def setup_routing(self):
        self.r_config = self.config.get('redis_manager', {})
        self.r_prefix = self.config['dispatcher_name']

        self.rules = []
        for rule in self.config.get('rules', []):
            if 'keyword' not in rule or 'app' not in rule:
                raise ConfigError("Rule definition %r must contain values for"
                                  " both 'app' and 'keyword'" % rule)
            rule = rule.copy()
            rule['keyword'] = rule['keyword'].lower()
            self.rules.append(rule)
        keyword_mappings = self.config.get('keyword_mappings', {})
        for transport_name, keyword in keyword_mappings.items():
            self.rules.append({'app': transport_name,
                               'keyword': keyword.lower()})
        self.fallback_application = self.config.get('fallback_application')
        self.transport_mappings = self.config['transport_mappings']
        self.expire_routing_timeout = int(self.config.get(
            'expire_routing_memory', self.DEFAULT_ROUTING_TIMEOUT))

        # FIXME: The following is a hack to deal with sync-only setup.
        self._redis_d = TxRedisManager.from_config(self.r_config)
        self._redis_d.addCallback(lambda m: m.sub_manager(self.r_prefix))
        self._redis_d.addCallback(self._setup_redis)

    def setup(self, redis=None, message_sender=None):
        if redis is None:
            redis = yield TxRedisManager.from_config(self.redis_config)

        if message_sender is None:
            message_sender = MessageSender(
                'amqp-spec-0-8.xml', self.amqp_config)

        self.redis = redis
        self.message_sender = message_sender
        self.message_sender.setServiceParent(self.service)

        self.inbounds = InboundMessageStore(
            self.redis, self.config.inbound_message_ttl)

        self.outbounds = OutboundMessageStore(
            self.redis, self.config.outbound_message_ttl)

        self.message_rate = MessageRateStore(self.redis)

        self.plugins = []
        for plugin_config in self.config.plugins:
            cls = load_class_by_string(plugin_config['type'])
            plugin = cls()
            yield plugin.start_plugin(plugin_config, self.config)
            self.plugins.append(plugin)

        yield Channel.start_all_channels(
            self.redis, self.config, self.service, self.plugins)

 def setup(self):
     self.r_config = self.config.get('redis_manager', {})
     self.keys_per_user_hard = self.config.get(
         'keys_per_user_hard', self.config.get('keys_per_user', 100))
     self.keys_per_user_soft = self.config.get(
         'keys_per_user_soft', int(0.8 * self.keys_per_user_hard))
     self.redis = yield TxRedisManager.from_config(self.r_config)

    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 setup_middleware(self):
     store_prefix = self.config.get('store_prefix', 'message_store')
     r_config = self.config.get('redis_manager', {})
     self.redis = yield TxRedisManager.from_config(r_config)
     manager = TxRiakManager.from_config(self.config.get('riak_manager'))
     self.store = MessageStore(manager,
                               self.redis.sub_manager(store_prefix))
     self.store_on_consume = self.config.get('store_on_consume', True)

 def setup(self):
     super(GoLoggingResource, self).setup()
     redis_config = self.config.get('redis_manager', {})
     max_logs_per_conversation = self.config.get(
         'max_logs_per_conversation')
     self._redis = yield TxRedisManager.from_config(redis_config)
     self.log_manager = LogManager(
         self._redis, max_logs_per_conversation=max_logs_per_conversation)

 def setup_middleware(self):
     store_prefix = self.config.store_prefix
     r_config = self.config.redis_manager
     self.redis = yield TxRedisManager.from_config(r_config)
     manager = TxRiakManager.from_config(self.config.riak_manager)
     self.store = MessageStore(manager,
                               self.redis.sub_manager(store_prefix))
     self.store_on_consume = self.config.store_on_consume

 def setup_transport(self):
     config = self.get_static_config()
     self.redis = yield TxRedisManager.from_config(config.redis_manager)
     self.retries = 0
     self.delay = config.initial_delay
     self.reconnect_call = None
     self.client = StreamingClient(self.agent_factory)
     self.connect_api_clients()

 def setup_worker(self):
     config = self.get_static_config()
     redis_manager = yield TxRedisManager.from_config(config.redis_manager)
     lop_manager = LopManager(redis_manager)
     rpc = LopApiServer(lop_manager)
     addIntrospection(rpc)
     site = Site(rpc)
     self.addService(strports.service(config.twisted_endpoint, site))

 def get_manager(self):
     manager = yield TxRedisManager.from_config({
         'FAKE_REDIS': 'yes',
         'key_prefix': 'redistest',
     })
     self.add_cleanup(self.cleanup_manager, manager)
     yield manager._purge_all()
     returnValue(manager)

 def setup_middleware(self):
     self.validate_config()
     self.metric_publisher = yield self.worker.start_publisher(MetricPublisher)
     # We don't use a VumiApi here because we don't have a Riak config for
     # it.
     self.redis = yield TxRedisManager.from_config(self.config["redis_manager"])
     self.metric_manager = MetricManager(self.manager_name + ".", publisher=self.metric_publisher)
     self.metric_manager.start_polling()

 def setUp(self):
     config = {
         'key_prefix': 'heartbeats',
         'db': 5,
         'FAKE_REDIS': True,
     }
     self.redis = yield TxRedisManager.from_config(config)
     self.add_cleanup(self.cleanup_redis)
     self.stg = storage.Storage(self.redis)

    def setup_application(self):
        self.redis = yield TxRedisManager.from_config(
            self.config['redis_manager'])

        self.inbounds = InboundMessageStore(
            self.redis, self.config['inbound_ttl'])

        self.outbounds = OutboundMessageStore(
            self.redis, self.config['outbound_ttl'])

 def setup_application(self):
     config = self.get_static_config()
     self.redis = None
     if config.allow_replies:
         self.redis = yield TxRedisManager.from_config(config.redis_manager)
     send_resource = self.get_protected_resource(SendResource(self))
     self.web_resource = yield self.start_web_resources(
         [(send_resource, config.web_path), (HealthResource(), "health")], config.web_port
     )

 def from_config_async(cls, config, command_publisher=None,
                       metric_publisher=None):
     # Note: This takes a publisher rather than a client to avoid leaking
     #       AMQP channels by making our own transient publishers.
     riak_config, redis_config = cls._parse_config(config)
     manager = TxRiakManager.from_config(riak_config)
     redis = yield TxRedisManager.from_config(redis_config)
     sender = AsyncMessageSender(command_publisher)
     returnValue(cls(manager, redis, sender, metric_publisher))

 def from_redis_config(cls, config, key_prefix=None,
                       max_session_length=None, gc_period=None):
     """Create a `SessionManager` instance using `TxRedisManager`.
     """
     from vumi.persist.txredis_manager import TxRedisManager
     d = TxRedisManager.from_config(config)
     if key_prefix is not None:
         d.addCallback(lambda m: m.sub_manager(key_prefix))
     return d.addCallback(lambda m: cls(m, max_session_length, gc_period))

    def setup_middleware(self):
        self.redis = yield TxRedisManager.from_config(
            self.config.redis_manager)
        self.timeout = self.config.timeout
        self.field_name = self.config.field_name
        self.clock = reactor

        namespace_type = self.config.namespace_type
        self.namespace_handler = getattr(
            self, self.NAMESPACE_HANDLERS[namespace_type])

    def setup_worker(self):
        config = self.get_static_config()
        self._riak = yield TxRiakManager.from_config(config.riak_manager)
        redis = yield TxRedisManager.from_config(config.redis_manager)
        self.store = MessageStore(self._riak, redis)

        site = build_web_site({
            config.web_path: MessageStoreResource(self.store),
            config.health_path: httprpc.HttpRpcHealthResource(self),
        })
        self.addService(
            StreamServerEndpointService(config.twisted_endpoint, site))

    def get_redis(self, **kwargs):
        from vumi.persist.txredis_manager import TxRedisManager
        # Fake redis
        fake_redis = yield TxRedisManager._fake_manager(
            FakeRedis(async=True, **kwargs), {
                "config": {},
                "key_prefix": 'redistest',
            })
        self.add_cleanup(fake_redis._close)
        # Real redis
        config = {
            'FAKE_REDIS': 'yes',
            'key_prefix': 'redistest',
        }
        config.update(kwargs)
        real_redis = yield TxRedisManager.from_config(config)
        self.add_cleanup(self.cleanup_manager, real_redis)
        # Both redises
        yield real_redis._purge_all()

        returnValue(RedisPairWrapper(self, fake_redis, real_redis))

    def startWorker(self):
        web_path = self.config['web_path']
        web_port = int(self.config['web_port'])
        health_path = self.config['health_path']

        riak = yield TxRiakManager.from_config(self.config['riak_manager'])
        redis = yield TxRedisManager.from_config(self.config['redis_manager'])
        self.store = MessageStore(riak, redis)

        self.webserver = self.start_web_resources([
            (MessageStoreAPI(self.store), web_path),
            (httprpc.HttpRpcHealthResource(self), health_path),
            ], web_port)