Stream a message payload as a series of frames. This allows control frames to be interleaved
between parts of the message.
508901
Sink newMessageSink(int formatOpcode, long contentLength) {
    if (activeWriter) {
      throw new IllegalStateException("Another message writer is active. Did you call close()?");
    }
    activeWriter = true;

    // Reset FrameSink state for a new writer.
    frameSink.formatOpcode = formatOpcode;
    frameSink.contentLength = contentLength;
    frameSink.isFirstFrame = true;
    frameSink.closed = false;

    return frameSink;
  }
596187
function deliverMessage (handle, topic, payload) {
  if (handle === 'broadcast') {
    const channel = ChannelsManager.resolve(topic)

    if (!channel) {
      return debug('broadcast topic %s cannot be handled by any channel', topic)
    }

    channel.clusterBroadcast(topic, payload)
    return
  }

  debug('dropping packet, since %s handle is not allowed', handle)
}
881716
private void sendPingFrame(ChannelHandlerContext ctx, SendPingCommand msg,
      ChannelPromise promise) {
    // Don't check lifecycleManager.getShutdownStatus() since we want to allow pings after shutdown
    // but before termination. After termination, messages will no longer arrive because the
    // pipeline clears all handlers on channel close.

    PingCallback callback = msg.callback();
    Executor executor = msg.executor();
    // we only allow one outstanding ping at a time, so just add the callback to
    // any outstanding operation
    if (ping != null) {
      promise.setSuccess();
      ping.addCallback(callback, executor);
      return;
    }

    // Use a new promise to prevent calling the callback twice on write failure: here and in
    // NettyClientTransport.ping(). It may appear strange, but it will behave the same as if
    // ping != null above.
    promise.setSuccess();
    promise = ctx().newPromise();
    // set outstanding operation
    long data = USER_PING_PAYLOAD;
    Stopwatch stopwatch = stopwatchFactory.get();
    stopwatch.start();
    ping = new Http2Ping(data, stopwatch);
    ping.addCallback(callback, executor);
    // and then write the ping
    encoder().writePing(ctx, false, USER_PING_PAYLOAD, promise);
    ctx.flush();
    final Http2Ping finalPing = ping;
    promise.addListener(new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isSuccess()) {
          transportTracer.reportKeepAliveSent();
        } else {
          Throwable cause = future.cause();
          if (cause instanceof ClosedChannelException) {
            cause = lifecycleManager.getShutdownThrowable();
            if (cause == null) {
              cause = Status.UNKNOWN.withDescription("Ping failed but for unknown reason.")
                  .withCause(future.cause()).asException();
            }
          }
          finalPing.failed(cause);
          if (ping == finalPing) {
            ping = null;
          }
        }
      }
    });
  }
564756
async def send(self, message: Union[str, bytes],
                   binary: bool=False,
                   compress: Optional[int]=None) -> None:
        """"""
        if isinstance(message, str):
            message = message.encode('utf-8')
        if binary:
            await self._send_frame(message, WSMsgType.BINARY, compress)
        else:
            await self._send_frame(message, WSMsgType.TEXT, compress)
952312
def send_frame(self, frame):
        

        # Get the socket
        sock = self.manager.sock

        if not sock:
            raise ValueError("UDPTendrilManager not running")

        # Send the packet
        try:
            sock.sendto(self._send_streamify(frame), self.remote_addr)
        except Exception:
            # We're a best-effort service anyway, so ignore exceptions
            pass
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)
620191
def send_frame(self, message, opcode):
        
        if self._closed:
            raise WebSocketError("Connection is already closed")
        if opcode == self.OPCODE_TEXT:
            message = self._encode_bytes(message)
        elif opcode == self.OPCODE_BINARY:
            message = six.binary_type(message)
        header = Header.encode_header(True, opcode, '', len(message), 0)
        try:
            self.stream.write(header + message)
        except socket_error:
            raise WebSocketError("Socket is dead")
167486
def recv_rpc(self, context, payload):
        """"""
        logger.debug("Adding RPC payload to ControlBuffer queue: %s", payload)
        self.buf.put(('rpc', (context, payload)))
        with self.cv:
            self.cv.notifyAll()
278482
def into_buffer(self, buf, offset):
    
    self.payload = containerize(exclude_fields(self))
    self.parser = MsgEphemerisGPSDepF._parser
    self.stream_payload.reset(buf, offset)
    return self.pack_into(buf, offset, self._build_payload)
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)
849690
private synchronized void processProgressEvent(DownstreamChannel channel, WrappedByteBuffer buffer) {
        LOG.entering(CLASS_NAME, "processProgressEvent", buffer);
        try {
            
            // update timestamp that is used to record the timestamp of last received message
            channel.lastMessageTimestamp.set(System.currentTimeMillis());
            channel.buffersToRead.add(buffer);

            WebSocketEmulatedDecoderListener<DownstreamChannel> decoderListener = new WebSocketEmulatedDecoderListener<DownstreamChannel>() {

                @Override
                public void messageDecoded(DownstreamChannel channel, WrappedByteBuffer message) {
                    processMessage(channel, message);
                }

                @Override
                public void messageDecoded(DownstreamChannel channel, String message) {
                    processMessage(channel, message);
                }

                @Override
                public void commandDecoded(DownstreamChannel channel, WrappedByteBuffer command) {
                    int commandByte = command.array()[0];
                    if (commandByte == 0x30 && command.array()[1] == 0x31) { //reconnect
                        // KG-5615: Set flag - but do not reconnect until request has loaded
                        LOG.fine("Reconnect command");
                        channel.reconnecting.set(true);
                    }
                    else if (commandByte == 0x30 && command.array()[1] == 0x32) {
                        channel.closing.set(true);

                        // Cancel the idle timer if running
                        stopIdleTimer(channel);
                        
                        //close frame received
                        int code = CloseCommandMessage.CLOSE_NO_STATUS;
                        String reason = null;
                        command.skip(2); //skip first 2 bytes 0x30, 0x32
                        if (command.hasRemaining()) {
                            code = command.getShort();
                        }
                        if (command.hasRemaining()) {
                            reason = command.getString(Charset.forName("UTF-8"));
                        }
                        CloseCommandMessage message = new CloseCommandMessage(code, reason);
                        listener.commandMessageReceived(channel, message);
                    }
                }

                @Override
                public void pingReceived(DownstreamChannel channel) {
                    listener.pingReceived(channel);
                }
            };

            // Prevent multiple threads from entering
            synchronized (channel.decoder) {
                channel.decoder.decode(channel, in, decoderListener);
            }
        } catch (Exception e) {
            LOG.log(Level.FINE, e.getMessage(), e);
            e.printStackTrace();
            
            listener.downstreamFailed(channel, e);
        }
    }
1246561
function FrameWriter(opcode, streamId) {
  if (!opcode) {
    throw new Error('Opcode not provided');
  }
  this.streamId = streamId;
  this.buffers = [];
  this.opcode = opcode;
}
1252236
def _process_method_frame(self, channel, payload):
        
        method_sig = unpack('>HH', payload[:4])
        args = AMQPReader(payload[4:])

        if method_sig in _CONTENT_METHODS:
            #
            # Save what we've got so far and wait for the content-header
            #
            self.partial_messages[channel] = _PartialMessage(method_sig, args)
            self.expected_types[channel] = 2
        else:
            self.queue.put((channel, method_sig, args, None))
910768
def encode(self, frame: Frame) -> Frame:
        
        # Skip control frames.
        if frame.opcode in CTRL_OPCODES:
            return frame

        # Since we always encode and never fragment messages, there's no logic
        # similar to decode() here at this time.

        if frame.opcode != OP_CONT:
            # Re-initialize per-message decoder.
            if self.local_no_context_takeover:
                self.encoder = zlib.compressobj(
                    wbits=-self.local_max_window_bits, **self.compress_settings
                )

        # Compress data frames.
        data = self.encoder.compress(frame.data) + self.encoder.flush(zlib.Z_SYNC_FLUSH)
        if frame.fin and data.endswith(_EMPTY_UNCOMPRESSED_BLOCK):
            data = data[:-4]

        # Allow garbage collection of the encoder if it won't be reused.
        if frame.fin and self.local_no_context_takeover:
            del self.encoder

        return frame._replace(data=data, rsv1=True)
197744
def encode(frame)
      frames = if frame[:type] == :headers || frame[:type] == :push_promise
        encode_headers(frame) # HEADERS and PUSH_PROMISE may create more than one frame
      else
        [frame]               # otherwise one frame
      end

      frames.map { |f| @framer.generate(f) }
    end
166992
def from_frame(cls, frame):
        """"""
        message = cls()

        # Check checksum first
        checksum = 0
        for i in range(0, len(frame) - 1):
            checksum += frame[i]

        if (checksum % 0x100) != frame[len(frame) - 1]:
            raise KNXException('Checksum error in frame {}, '
                               'expected {} but got {}'
                               .format(tohex(frame), frame[len(frame) - 1],
                                       checksum % 0x100))

        message.repeat = (frame[0] >> 5) & 0x01
        message.priority = (frame[0] >> 2) & 0x03
        message.src_addr = (frame[1] << 8) + frame[2]
        message.dst_addr = (frame[3] << 8) + frame[4]
        message.multicast = (frame[5] >> 7)
        message.routing = (frame[5] >> 4) & 0x03
        message.length = frame[5] & 0x0f
        message.data = frame[6:-1]

        if len(message.data) + 1 != message.length:
            raise KNXException(
                'Frame {} has not the correct length'.format(tohex(frame)))

        return message
382010
def encode_bridged_message(routing, header, payload, seq):
    

    # change header requester addresses for bridging
    header.rq_sa = routing[-1].rq_sa
    header.rs_sa = routing[-1].rs_sa
    tx_data = encode_ipmb_msg(header, payload)

    for bridge in reversed(routing[:-1]):
        tx_data = encode_send_message(tx_data,
                                      rq_sa=bridge.rq_sa,
                                      rs_sa=bridge.rs_sa,
                                      channel=bridge.channel,
                                      seq=seq)

    return tx_data