def handle_PUT_in_reply_to(self, request, payload, in_reply_to):
        user_account = request.getUser()
        conversation = yield self.get_conversation(user_account)

        reply_to = yield self.vumi_api.mdb.get_inbound_message(in_reply_to)
        if reply_to is None:
            self.client_error_response(request, 'Invalid in_reply_to value')
            return

        reply_to_mdh = MessageMetadataHelper(self.vumi_api, reply_to)
        try:
            msg_conversation_key = reply_to_mdh.get_conversation_key()
        except KeyError:
            log.warning('Invalid reply to message %r which has no conversation'
                        ' key' % (reply_to,))
            msg_conversation_key = None
        if msg_conversation_key != conversation.key:
            self.client_error_response(request, 'Invalid in_reply_to value')
            return

        msg_options = ReplyToOptions(payload)
        if not msg_options.is_valid:
            self.client_error_response(request, msg_options.error_msg)
            return

        continue_session = (msg_options.session_event
                            != TransportUserMessage.SESSION_CLOSE)
        helper_metadata = conversation.set_go_helper_metadata()
        helper_metadata.update(self.get_load_balancer_metadata(payload))

        msg = yield self.worker.reply_to(
            reply_to, msg_options.content, continue_session,
            helper_metadata=helper_metadata)

        self.successful_send_response(request, msg)

 def send_message_to_client(self, message, conversation, push_url):
     if push_url is None:
         log.warning(
             "push_message_url not configured for conversation: %s" % (
                 conversation.key))
         return
     return self.push(push_url, message)

    def _handle_receive(self, message_id, request):
        if not self._check_request_args(request, ["secret", "sent_timestamp", "sent_to", "from", "message"]):
            log.warning("Bad request: %r (args: %r)" % (request, request.args))
            yield self._send_response(message_id, success=self.SMSSYNC_FALSE)
            return
        msginfo = yield self.msginfo_for_request(request)
        supplied_secret = request.args["secret"][0]
        if msginfo is None or (msginfo.smssync_secret and not msginfo.smssync_secret == supplied_secret):
            log.warning("Bad secret or account: %r (args: %r)" % (request, request.args))
            yield self._send_response(message_id, success=self.SMSSYNC_FALSE)
            return

        timestamp = self._parse_timestamp(request)

        normalize = lambda raw: normalize_msisdn(raw, msginfo.country_code)
        message = {
            "message_id": message_id,
            "transport_type": self.transport_type,
            "to_addr": normalize(request.args["sent_to"][0]),
            "from_addr": normalize(request.args["from"][0]),
            "content": request.args["message"][0],
            "timestamp": timestamp,
        }
        self.add_msginfo_metadata(message, msginfo)
        yield self.publish_message(**message)
        self.callLater(
            self._reply_delay, self._respond_with_pending_messages, msginfo, message_id, success=self.SMSSYNC_TRUE
        )

    def process_command_send_message(self, user_account_key, conversation_key,
                                     command_data, **kwargs):
        if kwargs:
            log.info("Received unexpected command args: %s" % (kwargs,))
        conv = yield self.get_conversation(user_account_key, conversation_key)
        if conv is None:
            log.warning("Cannot find conversation '%s' for user '%s'." % (
                conversation_key, user_account_key))
            return

        log.info('Processing send_message: %s' % kwargs)
        to_addr = command_data['to_addr']
        content = command_data['content']
        msg_options = command_data['msg_options']
        in_reply_to = msg_options.pop('in_reply_to', None)
        self.add_conv_to_msg_options(conv, msg_options)
        if in_reply_to:
            msg = yield self.vumi_api.mdb.get_inbound_message(in_reply_to)
            if msg:
                yield self.reply_to(
                    msg, content,
                    helper_metadata=msg_options['helper_metadata'])
            else:
                log.warning('Unable to reply, message %s does not exist.' % (
                    in_reply_to))
        else:
            yield self.send_to(
                to_addr, content, endpoint='default', **msg_options)

    def push(self, url, vumi_message):
        config = self.get_static_config()
        data = vumi_message.to_json().encode('utf-8')
        try:
            auth, url = extract_auth_from_url(url.encode('utf-8'))
            headers = {
                'Content-Type': 'application/json; charset=utf-8',
            }
            if auth is not None:
                username, password = auth

                if username is None:
                    username = ''

                if password is None:
                    password = ''

                headers.update({
                    'Authorization': 'Basic %s' % (
                        base64.b64encode('%s:%s' % (username, password)),)
                })
            resp = yield http_request_full(
                url, data=data, headers=headers, timeout=config.timeout)
            if not (200 <= resp.code < 300):
                # We didn't get a 2xx response.
                log.warning('Got unexpected response code %s from %s' % (
                    resp.code, url))
        except SchemeNotSupported:
            log.warning('Unsupported scheme for URL: %s' % (url,))
        except HttpTimeoutError:
            log.warning("Timeout pushing message to %s" % (url,))
        except DNSLookupError:
            log.warning("DNS lookup error pushing message to %s" % (url,))
        except ConnectionRefusedError:
            log.warning("Connection refused pushing message to %s" % (url,))

    def handle_outbound_message(self, message):
        headers = {"Content-Type": "application/json; charset=utf-8"}
        headers.update(self.get_auth_headers())

        params = {
            "to_addr": message["to_addr"],
            "content": message["content"],
            "message_id": message["message_id"],
            "in_reply_to": message["in_reply_to"],
            "session_event": message["session_event"],
        }
        if "helper_metadata" in message:
            params["helper_metadata"] = message["helper_metadata"]

        resp = yield http_request_full(
            self.get_url("messages.json"),
            data=json.dumps(params).encode("utf-8"),
            headers=headers,
            method="PUT",
            agent_class=self.agent_factory,
        )

        if resp.code != http.OK:
            log.warning("Unexpected status code: %s, body: %s" % (resp.code, resp.delivered_body))
            yield self.publish_nack(message["message_id"], reason="Unexpected status code: %s" % (resp.code,))
            return

        remote_message = json.loads(resp.delivered_body)
        yield self.map_message_id(remote_message["message_id"], message["message_id"])
        yield self.publish_ack(user_message_id=message["message_id"], sent_message_id=remote_message["message_id"])

    def handle_outbound_message(self, message):
        headers = {
            'Content-Type': 'application/json; charset=utf-8',
        }
        headers.update(self.get_auth_headers())

        params = {
            'to_addr': message['to_addr'],
            'content': message['content'],
            'message_id': message['message_id'],
            'in_reply_to': message['in_reply_to'],
            'session_event': message['session_event']
        }
        if 'helper_metadata' in message:
            params['helper_metadata'] = message['helper_metadata']

        resp = yield http_request_full(
            self.get_url('messages.json'),
            data=json.dumps(params).encode('utf-8'),
            headers=headers,
            method='PUT')

        if resp.code != http.OK:
            log.warning('Unexpected status code: %s, body: %s' % (
                resp.code, resp.delivered_body))
            yield self.publish_nack(message['message_id'],
                                    reason='Unexpected status code: %s' % (
                                        resp.code,))
            return

        remote_message = json.loads(resp.delivered_body)
        yield self.map_message_id(
            remote_message['message_id'], message['message_id'])
        yield self.publish_ack(user_message_id=message['message_id'],
                               sent_message_id=remote_message['message_id'])

 def _handle_send(self, message_id, request):
     msginfo = yield self.msginfo_for_request(request)
     if msginfo is None:
         log.warning("Bad account: %r (args: %r)" % (request, request.args))
         yield self._send_response(message_id, success=self.SMSSYNC_FALSE)
         return
     yield self._respond_with_pending_messages(msginfo, message_id, task="send", secret=msginfo.smssync_secret)

    def handle_bind_receiver_resp(self, pdu):
        if not pdu_ok(pdu):
            log.warning('Unable to bind: %r' % (command_status(pdu),))
            self.transport.loseConnection()
            return

        self.state = self.BOUND_STATE_RX
        return self.on_smpp_bind(seq_no(pdu))

 def normalise_provider(self, provider):
     if provider not in self.provider_mappings:
         log.warning(
             "No mapping exists for provider '%s', "
             "using '%s' as a fallback" % (provider, provider,))
         return provider
     else:
         return self.provider_mappings[provider]

 def _handle_submit_sm_resp_callback(self, message_id, smpp_message_id,
                                     command_status, cb):
     if message_id is None:
         # We have no message_id, so log a warning instead of calling the
         # callback.
         log.warning("Failed to retrieve message id for deliver_sm_resp."
                     " ack/nack from %s discarded."
                     % self.vumi_transport.transport_name)
     else:
         return cb(message_id, smpp_message_id, command_status)

 def process_message_in_sandbox(self, msg):
     # TODO remove the delivery class inference and injection into the
     # message once we have message address types
     metadata = msg['helper_metadata']
     metadata['delivery_class'] = self.infer_delivery_class(msg)
     config = yield self.get_config(msg)
     if not config.javascript:
         log.warning("No JS for conversation: %s" % (
             config.conversation.key,))
         return
     yield super(JsBoxApplication, self).process_message_in_sandbox(msg)

    def on_unsupported_command_id(self, pdu):
        """
        Called when an SMPP PDU is received for which no handler function has
        been defined.

        :param dict pdu:
            The dict result one gets when calling ``smpp.pdu.unpack_pdu()``
            on the received PDU
        """
        log.warning(
            'Received unsupported SMPP command_id: %r' % (command_id(pdu),))

 def consume_user_message(self, message):
     msg_mdh = self.get_metadata_helper(message)
     conversation = yield msg_mdh.get_conversation()
     if conversation is None:
         log.warning("Cannot find conversation for message: %r" % (
             message,))
         return
     ignore = self.get_api_config(conversation, 'ignore_messages', False)
     if not ignore:
         push_url = self.get_api_config(conversation, 'push_message_url')
         yield self.send_message_to_client(message, conversation, push_url)

 def get_conversations(self, conv_pointers):
     results = yield gatherResults([
         self.get_conversation_fallback(account_key, conv_key)
         for account_key, conv_key in conv_pointers])
     conversations = []
     for pointer, conv in zip(conv_pointers, results):
         if conv is None:
             log.warning("Conversation %s for account %s not found." % (
                 pointer[1], pointer[0]))
         else:
             conversations.append(conv)
     returnValue(conversations)

    def handle_deliver_sm_ussd(self, pdu, pdu_params, pdu_opts):
        service_op = pdu_opts['ussd_service_op']
        mica_session_identifier = pdu_opts['user_message_reference']
        vumi_session_identifier = make_vumi_session_identifier(
            pdu_params['source_addr'], mica_session_identifier)

        session_event = self.ussd_service_op_map.get(service_op)

        if session_event == 'new':
            # PSSR request. Let's assume it means a new session.
            ussd_code = pdu_params['short_message']
            content = None

            yield self.session_manager.create_session(
                vumi_session_identifier, ussd_code=ussd_code)

        elif session_event == 'close':
            session = yield self.session_manager.load_session(
                vumi_session_identifier)
            ussd_code = session['ussd_code']
            content = None

            yield self.session_manager.clear_session(vumi_session_identifier)

        else:
            if session_event != 'continue':
                log.warning(('Received unknown %r ussd_service_op, '
                             'assuming continue.') % (service_op,))
                session_event = 'continue'

            session = yield self.session_manager.load_session(
                vumi_session_identifier)
            ussd_code = session['ussd_code']
            content = pdu_params['short_message']

        # This is stashed on the message and available when replying
        # with a `submit_sm`
        session_info = {
            'ussd_service_op': service_op,
            'session_identifier': mica_session_identifier,
        }

        decoded_msg = self.dcs_decode(content,
                                      pdu_params['data_coding'])

        result = yield self.handle_short_message_content(
            source_addr=pdu_params['source_addr'],
            destination_addr=ussd_code,
            short_message=decoded_msg,
            message_type='ussd',
            session_event=session_event,
            session_info=session_info)
        returnValue(result)

 def find_target(self, config, msg, connector_name):
     endpoint_name = msg.get_routing_endpoint()
     endpoint_routing = config.routing_table.get(connector_name)
     if endpoint_routing is None:
         log.warning("No routing information for connector '%s'" % (
                 connector_name,))
         return None
     target = endpoint_routing.get(endpoint_name)
     if target is None:
         log.warning("No routing information for endpoint '%s' on '%s'" % (
                 endpoint_name, connector_name,))
         return None
     return target

    def handle_delivery_report(self, receipted_message_id, delivery_status):
        message_id = yield self.message_stash.get_internal_message_id(
            receipted_message_id)
        if message_id is None:
            log.warning("Failed to retrieve message id for delivery report."
                        " Delivery report from %s discarded."
                        % self.transport_name)
            return

        dr = yield self.publish_delivery_report(
            user_message_id=message_id,
            delivery_status=delivery_status)
        returnValue(dr)

 def delivery_report(self, message_id, message_state):
     delivery_status = self.delivery_status(message_state)
     message_id = yield self.r_get_id_for_third_party_id(message_id)
     if message_id is None:
         log.warning("Failed to retrieve message id for delivery report."
                     " Delivery report from %s discarded."
                     % self.transport_name)
         return
     log.msg("PUBLISHING DELIV REPORT: %s %s" % (message_id,
                                                 delivery_status))
     returnValue((yield self.publish_delivery_report(
                 user_message_id=message_id,
                 delivery_status=delivery_status)))

    def handle_raw_inbound_message(self, msgid, request):
        request_body = request.content.read()
        log.msg("Inbound message: %r" % (request_body,))
        try:
            body = ET.fromstring(request_body)
        except:
            log.warning("Error parsing request XML: %s" % (request_body,))
            yield self.finish_request(msgid, "", code=400)
            return

        # We always get this.
        session_id = body.find('session_id').text

        status_elem = body.find('status')
        if status_elem is not None:
            # We have a status message. These are all variations on "cancel".
            yield self.handle_status_message(msgid, session_id)
            return

        page_id = body.find('page_id').text

        # They sometimes send us page_id=0 in the middle of a session.
        if page_id == '0' and body.find('mobile_number') is not None:
            # This is a new session.
            session = yield self.save_session(
                session_id,
                from_addr=body.find('mobile_number').text,
                to_addr=body.find('gate').text)  # ???
            session_event = TransportUserMessage.SESSION_NEW
        else:
            # This is an existing session.
            session = yield self.session_manager.load_session(session_id)
            if 'from_addr' not in session:
                # We have a missing or broken session.
                yield self.finish_request(msgid, "", code=400)
                return
            session_event = TransportUserMessage.SESSION_RESUME

        content = body.find('data').text

        transport_metadata = {'session_id': session_id}
        self.publish_message(
                message_id=msgid,
                content=content,
                to_addr=session['to_addr'],
                from_addr=session['from_addr'],
                session_event=session_event,
                transport_name=self.transport_name,
                transport_type=self.config.get('transport_type'),
                transport_metadata=transport_metadata,
                )