<s> from django . conf import settings <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> from redis import ConnectionPool as RedisConnectionPool <EOL> from redis import Redis <EOL> from redis . connection import UnixDomainSocketConnection , Connection <EOL> from redis . connection import DefaultParser <EOL> from . views import BaseSseView <EOL> import json <EOL> CONNECTION_KWARGS = getattr ( settings , '<STR_LIT>' , { } ) <EOL> DEFAULT_CHANNEL = getattr ( settings , '<STR_LIT>' , '<STR_LIT>' ) <EOL> class ConnectionPoolManager ( object ) : <EOL> pools = { } <EOL> @ classmethod <EOL> def key_for_kwargs ( cls , kwargs ) : <EOL> return "<STR_LIT::>" . join ( [ str ( v ) for v in kwargs . values ( ) ] ) <EOL> @ classmethod <EOL> def connection_pool ( cls , ** kwargs ) : <EOL> pool_key = cls . key_for_kwargs ( kwargs ) <EOL> if pool_key in cls . pools : <EOL> return cls . pools [ pool_key ] <EOL> location = kwargs . get ( '<STR_LIT:location>' , None ) <EOL> if not location : <EOL> raise ImproperlyConfigured ( "<STR_LIT>" ) <EOL> params = { <EOL> '<STR_LIT>' : Connection , <EOL> '<STR_LIT>' : kwargs . get ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> '<STR_LIT:password>' : kwargs . get ( '<STR_LIT:password>' , None ) , <EOL> } <EOL> if location . startswith ( "<STR_LIT>" ) : <EOL> params [ '<STR_LIT>' ] = UnixDomainSocketConnection <EOL> params [ '<STR_LIT:path>' ] = location [ <NUM_LIT:5> : ] <EOL> else : <EOL> try : <EOL> params [ '<STR_LIT:host>' ] , params [ '<STR_LIT:port>' ] = location . split ( "<STR_LIT::>" ) <EOL> params [ '<STR_LIT:port>' ] = int ( params [ '<STR_LIT:port>' ] ) <EOL> except ValueError : <EOL> raise ImproperlyConfigured ( "<STR_LIT>" ) <EOL> cls . pools [ pool_key ] = RedisConnectionPool ( ** params ) <EOL> return cls . pools [ pool_key ] <EOL> class RedisQueueView ( BaseSseView ) : <EOL> redis_channel = DEFAULT_CHANNEL <EOL> def iterator ( self ) : <EOL> connection = _connect ( ) <EOL> pubsub = connection . pubsub ( ) <EOL> pubsub . subscribe ( self . get_redis_channel ( ) ) <EOL> for message in pubsub . listen ( ) : <EOL> if message [ '<STR_LIT:type>' ] == '<STR_LIT:message>' : <EOL> event , data = json . loads ( message [ '<STR_LIT:data>' ] . decode ( '<STR_LIT:utf-8>' ) ) <EOL> self . sse . add_message ( event , data ) <EOL> yield <EOL> def get_redis_channel ( self ) : <EOL> return self . redis_channel <EOL> def _connect ( ) : <EOL> pool = ConnectionPoolManager . connection_pool ( ** CONNECTION_KWARGS ) <EOL> return Redis ( connection_pool = pool ) <EOL> def send_event ( event_name , data , channel = DEFAULT_CHANNEL ) : <EOL> connection = _connect ( ) <EOL> connection . publish ( channel , json . dumps ( [ event_name , data ] ) ) <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' ] </s>
