def render_(self, request):
     log.msg("Send request: %s" % (request,))
     request.setHeader("content-type", "application/json; charset=utf-8")
     d = self.application.handle_raw_outbound_message(request)
     d.addCallback(lambda msgs: self.finish_request(request, msgs))
     d.addErrback(lambda f: self.fail_request(request, f))
     return NOT_DONE_YET

    def handle_event(self, event, handler_config):
        sms_copy = handler_config['sms_copy']
        conversation_key = handler_config['conversation_key']

        account_key = event['account_key']
        from_addr = event['content']['from_addr']

        user_api = self.get_user_api(account_key)
        contact = yield self.find_contact(account_key, from_addr)

        if not contact:
            log.msg('Unable to find contact for %s' % (from_addr,))
            return

        content = (sms_copy['swahili'] if contact.extra['language'] == '2'
                    else sms_copy['english'])

        conversation = yield user_api.get_wrapped_conversation(
                                                conversation_key)

        yield conversation.dispatch_command(
            'send_message', account_key, conversation.key,
            command_data={
                'batch_id': conversation.batch.key,
                'to_addr': from_addr,
                'content': content,
                'msg_options': {},
            }
        )

    def handle_inbound_text_message(self, request, wc_msg):
        double_delivery = yield self.check_for_double_delivery(
            request, wc_msg.msg_id)
        if double_delivery:
            log.msg('WeChat double delivery of message: %s' % (wc_msg.msg_id,))
            return

        lock = yield self.mark_as_seen_recently(wc_msg.msg_id)
        if not lock:
            log.msg('Unable to get lock for message id: %s' % (wc_msg.msg_id,))
            return

        config = self.get_static_config()
        if config.embed_user_profile:
            user_profile = yield self.get_user_profile(wc_msg.from_user_name)
        else:
            user_profile = {}

        mask = yield self.get_addr_mask(wc_msg.from_user_name)
        msg = yield self.publish_message(
            content=wc_msg.content,
            from_addr=wc_msg.from_user_name,
            to_addr=self.mask_addr(wc_msg.to_user_name, mask),
            timestamp=datetime.fromtimestamp(int(wc_msg.create_time)),
            transport_type=self.transport_type,
            transport_metadata={
                'wechat': {
                    'FromUserName': wc_msg.from_user_name,
                    'ToUserName': wc_msg.to_user_name,
                    'MsgType': 'text',
                    'MsgId': wc_msg.msg_id,
                    'UserProfile': user_profile,
                }
            })
        returnValue(msg)

    def handle_raw_inbound_message(self, message_id, request):
        values, errors = self.get_field_values(request, self.EXPECTED_FIELDS)

        channel = values.get('channel')
        if channel is not None and channel not in self.CHANNEL_LOOKUP.values():
            errors['unsupported_channel'] = channel

        if errors:
            log.msg('Unhappy incoming message: %s' % (errors,))
            yield self.finish_request(message_id, json.dumps(errors),
                                      code=http.BAD_REQUEST)
            return

        self.emit("AppositTransport receiving inbound message from "
                  "%(from)s to %(to)s" % values)

        yield self.publish_message(
            transport_name=self.transport_name,
            message_id=message_id,
            content=values['content'],
            from_addr=values['from'],
            to_addr=values['to'],
            provider='apposit',
            transport_type=self.TRANSPORT_TYPE_LOOKUP[channel],
            transport_metadata={'apposit': {'isTest': values['isTest']}})

        yield self.finish_request(
            message_id, json.dumps({'message_id': message_id}))

    def deliver_sm(self, *args, **kwargs):
        message_type = kwargs.get('message_type', 'sms')
        message = {
            'message_id': kwargs['message_id'],
            'to_addr': kwargs['destination_addr'],
            'from_addr': kwargs['source_addr'],
            'content': kwargs['short_message'],
            'transport_type': message_type,
            'transport_metadata': {},
            }

        if message_type == 'ussd':
            session_event = {
                'new': TransportUserMessage.SESSION_NEW,
                'continue': TransportUserMessage.SESSION_RESUME,
                'close': TransportUserMessage.SESSION_CLOSE,
                }[kwargs['session_event']]
            message['session_event'] = session_event
            session_info = kwargs.get('session_info')
            message['transport_metadata']['session_info'] = session_info

        log.msg("PUBLISHING INBOUND: %s" % (message,))
        # TODO: This logs messages that fail to serialize to JSON
        #       Usually this happens when an SMPP message has content
        #       we can't decode (e.g. data_coding == 4). We should
        #       remove the try-except once we handle such messages
        #       better.
        return self.publish_message(**message).addErrback(log.err)

    def switch_to_counters(self, batch_id):
        """
        Actively switch a batch from the old ``zcard()`` based approach
        to the new ``redis.incr()`` counter based approach.
        """
        uses_counters = yield self.uses_counters(batch_id)
        if uses_counters:
            log.msg('Batch %r has already switched to counters.' % (
                batch_id,))
            return

        # NOTE:     Under high load this may result in the counter being off
        #           by a few. Considering this is a cache that is to be
        #           reconciled we're happy for that to be the case.
        inbound_count = yield self.count_inbound_message_keys(batch_id)
        outbound_count = yield self.count_outbound_message_keys(batch_id)

        # We do `*_count or None` because there's a chance of getting back
        # a None if this is a new batch that's not received any traffic yet.
        yield self.redis.set(self.inbound_count_key(batch_id),
                             inbound_count or 0)
        yield self.redis.set(self.outbound_count_key(batch_id),
                             outbound_count or 0)

        yield self.truncate_inbound_message_keys(batch_id)
        yield self.truncate_outbound_message_keys(batch_id)

    def handle_outbound_message(self, message):
        metadata = message['transport_metadata']['mtn_nigeria_ussd']

        try:
            self.validate_outbound_message(message)
        except CodedXmlOverTcpError as e:
            log.msg(e)
            yield self.publish_nack(message['message_id'], "%s" % e)
            yield self.factory.client.send_error_response(
                metadata.get('session_id'),
                message.payload.get('in_reply_to'),
                e.code)
            return

        log.msg(
            'MtnNigeriaUssdTransport sending outbound message: %s' % message)

        end_session = (
            message['session_event'] == TransportUserMessage.SESSION_CLOSE)
        yield self.send_response(
            message_id=message['message_id'],
            session_id=metadata['session_id'],
            request_id=metadata['requestId'],
            star_code=metadata['starCode'],
            client_id=metadata['clientId'],
            msisdn=message['to_addr'],
            user_data=message['content'].encode(self.ENCODING),
            end_session=end_session)

    def handle_raw_inbound_message(self, msgid, request):
        parts = request.path.split('/')
        session_id = parts[-2]
        session_method = parts[-1]
        session_event, session_handler_name, sends_json = (
            self.METHOD_TO_HANDLER.get(session_method,
                                       (TransportUserMessage.SESSION_NONE,
                                       "handle_infobip_error", False)))
        session_handler = getattr(self, session_handler_name)
        req_content = request.content.read()
        log.msg("Incoming message: %r" % (req_content,))
        if sends_json:
            try:
                req_data = json.loads(req_content)
            except ValueError:
                # send bad JSON to error handler
                session_handler = self.handle_infobip_error
                req_data = {"error": "Invalid JSON"}
        else:
            req_data = {}

        message_dict = yield session_handler(msgid, session_id, req_data)
        if message_dict is not None:
            transport_metadata = {'session_id': session_id}
            message_dict.setdefault("message_id", msgid)
            message_dict.setdefault("session_event", session_event)
            message_dict.setdefault("content", None)
            message_dict["transport_name"] = self.transport_name
            message_dict["transport_type"] = self.config.get('transport_type',
                                                             'ussd')
            message_dict["transport_metadata"] = transport_metadata
            self.publish_message(**message_dict)

 def stopWorker(self):
     log.msg("HeartBeat: stopping worker")
     if self._task:
         self._task.stop()
         self._task = None
         yield self._task_done
     self._redis.close_manager()

    def setup_transport(self):
        config = self.get_static_config()
        log.msg('Starting SMPP Transport for: %s' % (config.twisted_endpoint,))

        default_prefix = '%[email protected]%s' % (config.system_id,
                                    config.transport_name)
        redis_prefix = config.split_bind_prefix or default_prefix
        self.redis = (yield TxRedisManager.from_config(
            config.redis_manager)).sub_manager(redis_prefix)

        self.dr_processor = config.delivery_report_processor(
            self, config.delivery_report_processor_config)
        self.deliver_sm_processor = config.deliver_short_message_processor(
            self, config.deliver_short_message_processor_config)
        self.submit_sm_processor = config.submit_short_message_processor(
            self, config.submit_short_message_processor_config)

        self.sequence_generator = self.sequence_class(self.redis)
        self.message_stash = SmppMessageDataStash(self.redis, config)
        self.throttled = None
        self._throttled_message_ids = []
        self._unthrottle_delayedCall = None
        self.factory = self.factory_class(self)

        self.service = self.start_service(self.factory)

        self.tps_counter = 0
        self.tps_limit = config.mt_tps
        if config.mt_tps > 0:
            self.mt_tps_lc = LoopingCall(self.reset_mt_tps)
            self.mt_tps_lc.clock = self.clock
            self.mt_tps_lc.start(1, now=True)
        else:
            self.mt_tps_lc = None

    def handle_raw_inbound_message(self, request_id, request):
        values, errors = self.get_field_values(request, self.EXPECTED_FIELDS)
        if errors:
            log.msg('Unhappy incoming message: %r' % (errors,))
            self.finish_request(
                request_id, json.dumps(errors), code=http.BAD_REQUEST)
            return

        to_addr = values["ussdServiceCode"]
        from_addr = values["msisdn"]
        session_event = yield self.session_event_for_transaction(
            values["transactionId"])

        yield self.publish_message(
            message_id=request_id,
            content=values["ussdRequestString"],
            to_addr=to_addr,
            from_addr=from_addr,
            provider='dmark',
            session_event=session_event,
            transport_type=self.transport_type,
            transport_metadata={
                'dmark_ussd': {
                    'transaction_id': values['transactionId'],
                    'transaction_time': values['transactionTime'],
                    'creation_time': values['creationTime'],
                }
            })

 def finish_expired_request(self, request_id, request):
     """
     Called on requests that timed out.
     """
     del self._requests[request_id]
     log.msg('Timing out on response for %s' % request.session['from_addr'])
     request.deferred.callback(self.response_for_error())

 def reset_mt_tps(self):
     if self.throttled and self.need_mt_throttling():
         if not self.service.is_bound():
             # We don't have a bound SMPP connection, so try again later.
             log.msg("Can't stop throttling while unbound, trying later.")
             return
         self.reset_mt_throttle_counter()
         self.stop_throttling(quiet=True)

 def connectionMade(self):
     self.state = 'OPEN'
     log.msg('STATE: %s' % (self.state))
     seq = yield self.get_next_seq()
     pdu = self.BIND_PDU(seq, **self.bind_params)
     log.msg(pdu.get_obj())
     self.send_pdu(pdu)
     self.schedule_lose_connection(self.CONNECTED_STATE)

 def handle_enquire_link(self, pdu):
     if pdu['header']['command_status'] == 'ESME_ROK':
         log.msg("enquire_link OK")
         sequence_number = pdu['header']['sequence_number']
         pdu_resp = EnquireLinkResp(sequence_number)
         self.send_pdu(pdu_resp)
     else:
         log.msg("enquire_link NOT OK: %r" % (pdu,))

 def handle_outbound(self, config, msg, conn_name):
     log.msg("Processing outbound message: %s" % (msg,))
     user_id = msg['to_addr']
     session_event = msg['session_event']
     session_manager = yield self.session_manager(config)
     session = yield session_manager.load_session(user_id)
     if session and (session_event == TransportUserMessage.SESSION_CLOSE):
         yield session_manager.clear_session(user_id)
     yield self.publish_outbound(msg)

 def lose_unbound_connection(self, required_state):
     if self.state != required_state:
         log.msg('Breaking connection due to binding delay, %s != %s\n' % (
             self.state, required_state))
         self._lose_conn = None
         self.transport.loseConnection()
     else:
         log.msg('Successful bind: %s, cancelling bind timeout' % (
             self.state))

 def register_client(self, client):
     # We add our own Deferred to the client here because we only want to
     # fire it after we're finished with our own deregistration process.
     client.registration_d = Deferred()
     client_addr = client.getAddress()
     log.msg("Registering client connected from %r" % client_addr)
     self._clients[client_addr] = client
     self.send_inbound_message(client, None,
                               TransportUserMessage.SESSION_NEW)

 def login(self):
     params = [
         ('requestId', self.gen_request_id()),
         ('userName', self.username),
         ('passWord', self.password),  # plaintext passwords, yay :/
         ('applicationId', self.application_id),
     ]
     self.send_packet(self.gen_session_id(), 'AUTHRequest', params)
     log.msg('Logging in')

    def startWorker(self):
        log.msg('Starting a %s dispatcher with config: %s'
                % (self.__class__.__name__, self.config))

        yield self.setup_endpoints()
        yield self.setup_router()
        yield self.setup_transport_publishers()
        yield self.setup_exposed_publishers()
        yield self.setup_transport_consumers()
        yield self.setup_exposed_consumers()