NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)
}

// Call the remove with the request and
func (h *StreamTypeHandler[Payload, Req, Resp]) Call(ctx context.Context, c Streamer, payload Payload) (st *TypedStream[Req, Resp], err error) {
	if c == nil {
		return nil, ErrDisconnected
	}
	var payloadB []byte
	if h.WithPayload {
		var err error

		progressTicker.Stop()
	}

	// Whatever drain the payloadCh to prevent from memory leaking.
	for len(writer.payloadCh) > 0 {
		payload := <-writer.payloadCh
		bufPool.Put(payload)
	}
}

// Sends a single whole record.
func (writer *messageWriter) SendRecord(payload *bytes.Buffer) error {
	select {
	case writer.payloadCh <- payload:
		return nil
	case <-writer.doneCh:

	var sub *subHandlerID
	if m.Payload == nil {
		m.Payload = GetByteBuffer()[:0]
	}
	h, err := m.UnmarshalMsg(b)
	if err != nil {
		return nil, nil, fmt.Errorf("read write: %v", err)
	}
	if len(m.Payload) == 0 && m.Flags&FlagPayloadIsZero == 0 {
		PutByteBuffer(m.Payload)
		m.Payload = nil
	}
	if m.Flags&FlagCRCxxh3 != 0 {
		const hashLen = 4

	errFatal(local.RegisterSingleHandler(handlerTest, func(payload []byte) ([]byte, *RemoteErr) {
		t.Log("1: server payload: ", len(payload), "bytes.")
		return append([]byte{}, payload...), nil
	}))
	// 2: Return as error
	errFatal(local.RegisterSingleHandler(handlerTest2, func(payload []byte) ([]byte, *RemoteErr) {
		t.Log("2: server payload: ", len(payload), "bytes.")
		err := RemoteErr(payload)
		return nil, &err
	}))

				fmt.Println("muxServer: Mux", m.ID, "send EOF", hErr)
			}
			msg.Flags |= FlagEOF
			if hErr != nil {
				msg.Flags |= FlagPayloadIsErr
				msg.Payload = []byte(*hErr)
			}
			msg.setZeroPayloadFlag()
			m.send(msg)
			return
		}
		msg.Payload = payload
		msg.setZeroPayloadFlag()
		m.send(msg)
	}
}

// handleRequests will handle the requests from the client and call the handler function.

	b.Cleanup(grid.Cleanup)
	// Create n managers.
	for _, remote := range grid.Managers {
		// Register a single handler which echos the payload.
		errFatal(remote.RegisterSingleHandler(handlerTest, func(payload []byte) ([]byte, *RemoteErr) {
			defer PutByteBuffer(payload)
			return append(GetByteBuffer()[:0], payload...), nil
		}))
		errFatal(rpc.Register(remote, func(req *testRequest) (resp *testResponse, err *RemoteErr) {
			return &testResponse{

  /** Send a ping with the supplied [payload]. */
  @Throws(IOException::class)
  fun writePing(payload: ByteString) {
    writeControlFrame(OPCODE_CONTROL_PING, payload)
  }

  /** Send a pong with the supplied [payload]. */
  @Throws(IOException::class)
  fun writePong(payload: ByteString) {
    writeControlFrame(OPCODE_CONTROL_PONG, payload)
  }

  /**

  }

  fun assertTextMessage(payload: String?) {
    assertThat(nextEvent()).isEqualTo(Message(string = payload))
  }

  fun assertBinaryMessage(payload: ByteString?) {
    assertThat(nextEvent()).isEqualTo(Message(payload))
  }

  fun assertPing(payload: ByteString) {
    assertThat(nextEvent()).isEqualTo(Ping(payload))
  }

  fun assertPong(payload: ByteString) {

    Class<?>[] groups() default {};

    Class<? extends Payload>[] payload() default {};

func (c *Connection) sendMsg(conn net.Conn, msg message, payload msgp.MarshalSizer) error {
	if payload != nil {
		if sz := payload.Msgsize(); cap(msg.Payload) < sz {
			PutByteBuffer(msg.Payload)
			msg.Payload = GetByteBufferCap(sz)[:0]
		}
		var err error
		msg.Payload, err = payload.MarshalMsg(msg.Payload)
		if err != nil {
			return err
		}
		defer PutByteBuffer(msg.Payload)
	}
	dst := GetByteBufferCap(msg.Msgsize())