Construct a gzip-compressed message containing multiple messages

    The given messages will be encoded, compressed, and sent as a single atomic
    message to Kafka.

    :param list message_set: a list of :class:`Message` instances
176095
def create_gzip_message(message_set):
    
    encoded_message_set = KafkaCodec._encode_message_set(message_set)

    gzipped = gzip_encode(encoded_message_set)
    return Message(0, CODEC_GZIP, None, gzipped)
436028
def make_message(self, data):

        

        data = self.codec.loads(data)
        msg = Message(
            data.get("data"),
            *data.get("args",[]),
            **data.get("kwargs",{})
        )
        msg.meta.update(data.get("meta"))
        self.trigger("make_message", data, msg)
        return msg
1019481
def new_message(cls, from_user, to_users, subject, content):
        
        thread = Thread.objects.create(subject=subject)
        for user in to_users:
            thread.userthread_set.create(user=user, deleted=False, unread=True)
        thread.userthread_set.create(user=from_user, deleted=True, unread=False)
        msg = cls.objects.create(thread=thread, sender=from_user, content=content)
        message_sent.send(sender=cls, message=msg, thread=thread, reply=False)
        return msg
1028754
def ts_stream_keys(self, table, timeout=None):
        
        msg_code = riak.pb.messages.MSG_CODE_TS_LIST_KEYS_REQ
        codec = self._get_codec(msg_code)
        msg = codec.encode_timeseries_listkeysreq(table, timeout)
        self._send_msg(msg.msg_code, msg.data)
        return PbufTsKeyStream(self, codec, self._ts_convert_timestamp)
788739
def unpack_messages(msgs):
        import msgpack
        

        for key, msg in msgs:
            record = msgpack.unpackb(msg)
            record['_key'] = key
            yield record
627742
def pack_msg(method, msg, pickle_protocol=PICKLE_PROTOCOL):
    """"""
    dump = io.BytesIO()
    pickle.dump(msg, dump, pickle_protocol)
    size = dump.tell()
    return (struct.pack(METHOD_STRUCT_FORMAT, method) +
            struct.pack(SIZE_STRUCT_FORMAT, size) + dump.getvalue())
333458
public void pushMessages(List<byte[]> messages) {
    for (byte[] message: messages) {
      this.futures.offer(producer.send(new ProducerRecord<>(topic, message), (recordMetadata, e) -> {
        if (e != null) {
          log.error("Failed to send message to topic {} due to exception: ", topic, e);
        }
      }));
    }

    //Once the low watermark of numFuturesToBuffer is hit, start flushing messages from the futures
    // buffer. In order to avoid blocking on newest messages added to futures queue, we only invoke future.get() on
    // the oldest messages in the futures buffer. The number of messages to flush is same as the number of messages added
    // in the current call. Note this does not completely avoid calling future.get() on the newer messages e.g. when
    // multiple threads enter the if{} block concurrently, and invoke flush().
    if (this.futures.size() >= this.numFuturesToBuffer) {
      flush(messages.size());
    }
  }
299736
def mget(self, *keys):
        
        return list(map(
            self._loads, self._client.hmget(self.key_prefix, *keys)))
981787
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1704")
  public Set<String> getAdvertisedMessageEncodings() {
    Set<String> advertisedDecompressors = new HashSet<>(decompressors.size());
    for (Entry<String, DecompressorInfo> entry : decompressors.entrySet()) {
      if (entry.getValue().advertised) {
        advertisedDecompressors.add(entry.getKey());
      }
    }
    return Collections.unmodifiableSet(advertisedDecompressors);
  }
537071
def buildMessage(headers, parts):
    
    message = multipart.MIMEMultipart('alternative')

    for name, value in headers.iteritems():
        name = name.title()
        if name == "From":
            multipart[name] = _encodeAddress(value)
        elif name in ["To", "Cc", "Bcc"]:
            multipart[name] = _encodeAddresses(value)
        else:
            multipart[name] = _encodeHeader(value)

    for partType, part in parts.iteritems():
        mimeText = text.MIMEText(part.encode("utf-8"), partType, "UTF-8")
        message.attach(mimeText.encode())

    return message
673275
def encode_offset_commit_request(cls, client_id, correlation_id,
                                     group, payloads):
        
        grouped_payloads = group_by_topic_and_partition(payloads)

        message = []
        message.append(cls._encode_message_header(client_id, correlation_id,
                                                  KafkaProtocol.OFFSET_COMMIT_KEY))
        message.append(write_short_string(group))
        message.append(struct.pack('>i', len(grouped_payloads)))

        for topic, topic_payloads in grouped_payloads.items():
            message.append(write_short_string(topic))
            message.append(struct.pack('>i', len(topic_payloads)))

            for partition, payload in topic_payloads.items():
                message.append(struct.pack('>iq', partition, payload.offset))
                message.append(write_short_string(payload.metadata))

        msg = b''.join(message)
        return struct.pack('>i%ds' % len(msg), len(msg), msg)
354313
def send(self, message, fragment_size=None, mask=False):
        
        for frame in self.message_to_frames(message, fragment_size, mask):
            self.send_frame(frame)
1187249
def incoming_messages(self) -> t.List[t.Tuple[float, bytes]]:
        
        approximate_messages = self._receive_buffer.qsize()
        messages = []
        for _ in range(approximate_messages):
            try:
                messages.append(self._receive_buffer.get_nowait())
            except queue.Empty:
                break
        return messages
1289414
def decompress_messages(self, partitions_offmsgs):
        

        for pomsg in partitions_offmsgs:
            if pomsg['message']:
                pomsg['message'] = self.decompress_fun(pomsg['message'])
            yield pomsg
606677
def unpack_messages(self, partitions_msgs):
        

        for pmsg in partitions_msgs:
            key = pmsg['_key']
            partition = pmsg['partition']
            offset = pmsg['offset']
            msg = pmsg.pop('message')
            if msg:
                try:
                    record = self.deserialize_fun(msg)
                except Exception as e:
                    log.error('Error unpacking record at partition:offset {}:{} (key: {} : {})'.format(partition, offset, key, repr(e)))
                    continue
                else:
                    if isinstance(record, dict):
                        pmsg['record'] = record
                        yield pmsg
                    else:
                        log.info('Record {} has wrong type'.format(key))
            else:
                yield pmsg
774950
public void pushMessages(List<Pair<K, V>> messages) {
    for (Pair<K, V> message: messages) {
      this.futures.offer(this.producer.send(new ProducerRecord<>(topic, message.getKey(), message.getValue()), (recordMetadata, e) -> {
        if (e != null) {
          log.error("Failed to send message to topic {} due to exception: ", topic, e);
        }
      }));
    }

    //Once the low watermark of numFuturesToBuffer is hit, start flushing messages from the futures
    // buffer. In order to avoid blocking on newest messages added to futures queue, we only invoke future.get() on
    // the oldest messages in the futures buffer. The number of messages to flush is same as the number of messages added
    // in the current call. Note this does not completely avoid calling future.get() on the newer messages e.g. when
    // multiple threads enter the if{} block concurrently, and invoke flush().
    if (this.futures.size() >= this.numFuturesToBuffer) {
      flush(messages.size());
    }
  }
719239
def send_many(kwargs_list):
    
    emails = []
    for kwargs in kwargs_list:
        emails.append(send(commit=False, **kwargs))
    Email.objects.bulk_create(emails)