Send a batch of messages

        :param str topic: a kafka topic
        :param ksr.transport.Message kmsgs: Messages to serialize
        :param int timeout: Timeout in seconds
        :return: Execution result
        :rtype: kser.result.Result
1174
def bulk_send(self, topic, kmsgs, timeout=60):
        

        try:
            for kmsg in kmsgs:
                self.client.send(
                    topic, self._onmessage(kmsg).dumps().encode("UTF-8")
                )
            self.client.flush(timeout=timeout)
            return Result(stdout="{} message(s) sent".format(len(kmsgs)))

        except Exception as exc:
            return Result.from_exception(exc)
38149
def returner(ret):
    '''
    
    '''
    if __salt__['config.option']('returner.kafka.topic'):
        topic = __salt__['config.option']('returner.kafka.topic')

        conn = _get_conn()
        producer = Producer({'bootstrap.servers': conn})
        producer.poll(0)
        producer.produce(topic, salt.utils.json.dumps(ret), str(ret).encode('utf-8'), callback=_delivery_report)

        producer.flush()
    else:
        log.error('Unable to find kafka returner config option: topic')
364847
def _send_to_timeseries(self, message):
        
        logging.debug("MESSAGE=" + str(message))

        result = None
        try:
            ws = self._get_websocket()
            ws.send(json.dumps(message))
            result = ws.recv()
        except (websocket.WebSocketConnectionClosedException, Exception) as e:
            logging.debug("Connection failed, will try again.")
            logging.debug(e)

            ws = self._get_websocket(reuse=False)
            ws.send(json.dumps(message))
            result = ws.recv()

        logging.debug("RESULT=" + str(result))
        return result
961792
async def send(self, message_type, message_content, timeout=None):
        
        return await self._sender.send(
            message_type, message_content, timeout=timeout)
266166
def _produce_request(self, node_id, acks, timeout, batches):
        
        produce_records_by_partition = collections.defaultdict(dict)
        for batch in batches:
            topic = batch.topic_partition.topic
            partition = batch.topic_partition.partition

            buf = batch.records.buffer()
            produce_records_by_partition[topic][partition] = buf

        kwargs = {}
        if self.config['api_version'] >= (0, 11):
            version = 3
            kwargs = dict(transactional_id=None)
        elif self.config['api_version'] >= (0, 10):
            version = 2
        elif self.config['api_version'] == (0, 9):
            version = 1
        else:
            version = 0
        return ProduceRequest[version](
            required_acks=acks,
            timeout=timeout,
            topics=[(topic, list(partition_info.items()))
                    for topic, partition_info
                    in six.iteritems(produce_records_by_partition)],
            **kwargs
        )
770823
def send(self, body, content_type='application/ubjson', timeout=5):
        
        if self.__send_ready.wait(timeout):
            try:
                with self.__send_lock:
                    # access denied response might be received inside send thread rather than here how to best handle?
                    self.__send_channel.basic_publish(msg=Message(body, delivery_mode=2, content_type=content_type),
                                                      exchange=self.__epid)
            except exceptions.AccessRefused as exc:
                raise_from(LinkException('Access denied'), exc)
            except (exceptions.AMQPError, SocketError) as exc:
                raise_from(LinkException('amqp/transport failure'), exc)
            except Exception as exc:  # pylint: disable=broad-except
                raise_from(LinkException('unexpected failure'), exc)
        else:
            exc = self.__send_exc
            if exc:
                raise_from(LinkException('Sender unavailable'), exc)
            else:
                raise LinkException('Sender unavailable (unknown error)')
562358
def sendmess(self, msgtype, payload, flags=0, size=0, offset=0, timeout=0):
        

        flags |= self.flags
        assert not (flags & FLG_PERSISTENCE)

        with self._new_connection() as conn:
            ret, _, data = conn.req(
                msgtype, payload, flags, size, offset, timeout)

        return ret, data
450314
def mpub(self, topic, *messages):
        ''''''
        with self.random_connection() as client:
            client.mpub(topic, *messages)
            return self.wait_response()
307780
def send_message(host, data, timeout=None, properties=None):
    
    channel = _get_channel(host, timeout)

    if not properties:
        properties = pika.BasicProperties(
            content_type="application/json",
            delivery_mode=2,
            headers={"UUID": str(uuid.uuid4())}
        )

    parameters = settings.get_amqp_settings()[host]

    channel.basic_publish(
        exchange=parameters["exchange"],
        routing_key=parameters["in_key"],
        properties=properties,
        body=data
    )
619175
def publish(self,topic,options=None,args=None,kwargs=None):
        
        topic = self.get_full_uri(topic)
        if options is None:
            options = {'acknowledge':True}
        if options.get('acknowledge'):
            request = PUBLISH(
                        options=options or {},
                        topic=topic,
                        args=args or [],
                        kwargs=kwargs or {}
                      )
            result = self.send_and_await_response(request)
            return result
        else:
            request = PUBLISH(
                        options=options or {},
                        topic=topic,
                        args=args or [],
                        kwargs=kwargs or {}
                      )
            self.send_message(request)
            return request.request_id
889865
def _SendOp(self):
    """"""
    msg = self._sender_queue.get()
    msgs = []
    background_msgs = []
    if not msg.require_fastpoll:
      background_msgs.append(msg)
    else:
      msgs.append(msg)

    count = 1
    size = len(msg.SerializeToString())

    while count < 100 and size < 1024 * 1024:
      try:
        msg = self._sender_queue.get(timeout=1)
        if not msg.require_fastpoll:
          background_msgs.append(msg)
        else:
          msgs.append(msg)
        count += 1
        size += len(msg.SerializeToString())
      except queue.Empty:
        break

    if msgs:
      self._SendMessages(msgs)
    if background_msgs:
      self._SendMessages(background_msgs, background=True)
496318
def publish(self, topic, payload, qos=0, retain=False):
        
        # TODO: check (full transmission) success
        return self._client.publish(topic, payload, qos, retain)
918673
def _send_and_wait(self, **kwargs):
        
        frame_id = self.next_frame_id
        kwargs.update(dict(frame_id=frame_id))
        self._send(**kwargs)
        timeout = datetime.now() + const.RX_TIMEOUT
        while datetime.now() < timeout:
            try:
                frame = self._rx_frames.pop(frame_id)
                raise_if_error(frame)
                return frame
            except KeyError:
                sleep(0.1)
                continue
        _LOGGER.exception(
            "Did not receive response within configured timeout period.")
        raise exceptions.ZigBeeResponseTimeout()
927388
def timeout_message(self, message):
        '''
        
        '''
        future = self.send_future_map.pop(message, None)
        # In a race condition the message might have been sent by the time
        # we're timing it out. Make sure the future is not None
        if future is not None:
            del self.send_timeout_map[message]
            if future.attempts < future.tries:
                future.attempts += 1
                log.debug('SaltReqTimeoutError, retrying. (%s/%s)', future.attempts, future.tries)
                self.send(
                    message,
                    timeout=future.timeout,
                    tries=future.tries,
                    future=future,
                )

            else:
                future.set_exception(SaltReqTimeoutError('Message timed out'))
709125
public KafkaMessage consume(String topic, long waitTime, TimeUnit waitTimeUnit) {
        BlockingQueue<ConsumerRecord<String, byte[]>> buffer = _getBuffer(topic);
        ConsumerRecord<String, byte[]> cr = buffer.poll();
        if (cr == null) {
            _fetch(buffer, topic, waitTime, waitTimeUnit);
            cr = buffer.poll();
        }
        return cr != null ? new KafkaMessage(cr).consumerGroupId(consumerGroupId) : null;
    }
357802
def set_room_topic(self, room_id, topic, timestamp=None):
        
        body = {
            "topic": topic
        }
        return self.send_state_event(room_id, "m.room.topic", body, timestamp=timestamp)
420614
public void send(MessageBuilder builder, SendMessageCallback cb, long timeout, TimeUnit unit) {
        Message msg = builder.build(this.messageBodyConverter);
        final String topic = msg.getTopic();
        MessageProducer producer = this.getOrCreateProducer(topic);
        producer.sendMessage(msg, cb, timeout, unit);
    }