Subscribe to the pub IPC
        and publish the messages
        on the right transport.
178952
def _setup_ipc(self):
        '''
        
        '''
        self.ctx = zmq.Context()
        log.debug('Setting up the %s publisher subscriber #%d', self._transport_type, self.pub_id)
        self.sub = self.ctx.socket(zmq.SUB)
        self.sub.connect(PUB_IPC_URL)
        self.sub.setsockopt(zmq.SUBSCRIBE, b'')
        try:
            self.sub.setsockopt(zmq.HWM, self.opts['hwm'])
            # zmq 2
        except AttributeError:
            # zmq 3
            self.sub.setsockopt(zmq.RCVHWM, self.opts['hwm'])
928342
def _setup_transport(self):
        '''
        
        '''
        if 'RAW' in self.error_whitelist:
            log.info('%s %d will publish partially parsed messages', self._transport_type, self.pub_id)
        if 'UNKNOWN' in self.error_whitelist:
            log.info('%s %d will publish unknown messages', self._transport_type, self.pub_id)
        transport_class = get_transport(self._transport_type)
        log.debug('Serializing the object for %s using %s',
                  self._transport_type,
                  self.serializer)
        self.serializer_fun = get_serializer(self.serializer)
        self.transport = transport_class(self.address,
                                         self.port,
                                         **self.publisher_opts)
        self.__transport_encrypt = True
        if hasattr(self.transport, 'NO_ENCRYPT') and\
           getattr(self.transport, 'NO_ENCRYPT') is True:
            self.__transport_encrypt = False
76011
def build_publish_pkt(self, mid, topic, payload, qos, retain, dup):
        """"""
        pkt = MqttPkt()
        payloadlen = len(payload)
        packetlen = 2 + len(topic) + payloadlen
        
        if qos > 0:
            packetlen += 2
        
        pkt.mid = mid
        pkt.command = NC.CMD_PUBLISH | ((dup & 0x1) << 3) | (qos << 1) | retain
        pkt.remaining_length = packetlen
        
        ret = pkt.alloc()
        if ret != NC.ERR_SUCCESS:
            return ret, None
        
        #variable header : Topic String
        pkt.write_string(topic)
        
        if qos > 0:
            pkt.write_uint16(mid)
        
        #payloadlen
        if payloadlen > 0:
            pkt.write_bytes(payload, payloadlen)
        
        return NC.ERR_SUCCESS, pkt
1174297
def connect_pub(self, timeout=None):
        '''
        
        '''
        if self.cpub:
            return True

        if self._run_io_loop_sync:
            with salt.utils.asynchronous.current_ioloop(self.io_loop):
                if self.subscriber is None:
                    self.subscriber = salt.transport.ipc.IPCMessageSubscriber(
                    self.puburi,
                    io_loop=self.io_loop
                )
                try:
                    self.io_loop.run_sync(
                        lambda: self.subscriber.connect(timeout=timeout))
                    self.cpub = True
                except Exception:
                    pass
        else:
            if self.subscriber is None:
                self.subscriber = salt.transport.ipc.IPCMessageSubscriber(
                self.puburi,
                io_loop=self.io_loop
            )

            # For the asynchronous case, the connect will be defered to when
            # set_event_handler() is invoked.
            self.cpub = True
        return self.cpub
1245985
def publish(self, topic, dct):
        
        get_logger().info("Publishing message {} on routing key "
                          "{}...".format(dct, topic))

        self._channel.basic_publish(
            exchange=self.exchange,
            routing_key=topic,
            body=json.dumps(dct)
        )
3064
def subscribe(self, receiver):
        ""
        log.debug('{0}.{1} subscribe to {2}'
                  .format(receiver.__module__, receiver.__name__, self))
        self.signal.connect(receiver)
1197135
def onMessage(self, payload, is_binary):
        
        if not is_binary:
            payload = payload.decode('utf-8')

            logger.debug("Incoming message ({peer}) : {message}".format(
                peer=self.peer, message=payload))

            # Publish ON_RECEIVE message
            self.factory.mease.publisher.publish(
                message_type=ON_RECEIVE,
                client_id=self._client_id,
                client_storage=self.storage,
                message=payload)
879458
def subscribe(self, destination, id, ack='auto', headers=None, **keyword_headers):
        
        self.transport.subscriptions[id] = destination
1038614
def _publish_queue_grpc(self):
        
        messages = EventHub_pb2.Messages(msg=self._tx_queue)
        publish_request = EventHub_pb2.PublishRequest(messages=messages)
        self.grpc_manager.send_message(publish_request)
258101
def publish(self, **kwargs):
        ""
        log.debug('Publish to {0}'.format(self))
        self.signal.send(sender=self.name, **kwargs)
378668
def publish(self, msg):
        '''
        
        '''
        if not self.streams:
            return

        pack = salt.transport.frame.frame_msg_ipc(msg, raw_body=True)

        for stream in self.streams:
            self.io_loop.spawn_callback(self._write, stream, pack)
869826
async def rpc_message(self, request, message):
        ''''''
        await self.pubsub.publish(self.channel, message)
        return 'OK'
577977
def listen(self):
        ''''''
        try:
            self._pubsub.subscribe(self._channels)
            for message in self._pubsub.listen():
                if message['type'] == 'message':
                    yield message
        finally:
            self._channels = []
710056
def set_pubsub_channels(self, request, channels):
        
        facility = request.path_info.replace(settings.WEBSOCKET_URL, '', 1)

        # initialize publishers
        audience = {
            'users': 'publish-user' in channels and [SELF] or [],
            'groups': 'publish-group' in channels and [SELF] or [],
            'sessions': 'publish-session' in channels and [SELF] or [],
            'broadcast': 'publish-broadcast' in channels,
        }
        self._publishers = set()
        for key in self._get_message_channels(request=request, facility=facility, **audience):
            self._publishers.add(key)

        # initialize subscribers
        audience = {
            'users': 'subscribe-user' in channels and [SELF] or [],
            'groups': 'subscribe-group' in channels and [SELF] or [],
            'sessions': 'subscribe-session' in channels and [SELF] or [],
            'broadcast': 'subscribe-broadcast' in channels,
        }
        self._subscription = self._connection.pubsub()
        for key in self._get_message_channels(request=request, facility=facility, **audience):
            self._subscription.subscribe(key)
949189
def _publish(self, msg):
        
        connection = self._connection.clone()
        publish = connection.ensure(self.producer, self.producer.publish,
                                    errback=self.__error_callback,
                                    max_retries=MQ_PRODUCER_MAX_RETRIES)
        publish(json.dumps(msg), exchange=self._exchange,
                routing_key=self._routing_key, declare=[self._queue])
        logging.debug('Publisher: message sent: %s', msg)
248971
def multipublish_tcp(self, topic, messages, **kwargs):
        
        return self.__tcp_client.multipublish(topic, messages, **kwargs)
474263
def listen(self):
        
        for message in self.pubsub.listen():
            if message['type'] == 'message':
                message_type, client_id, client_storage, args, kwargs = self.unpack(
                    message['data'])

                self.dispatch_message(
                    message_type, client_id, client_storage, args, kwargs)