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

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())

				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.

		}
	}
	payload := make([]byte, 1, plSize)
	payload[0] = xlMetaInlineDataVer
	payload = msgp.AppendMapHeader(payload, uint32(len(keys)))
	for i := range keys {
		payload = msgp.AppendStringFromBytes(payload, keys[i])
		payload = msgp.AppendBytes(payload, vals[i])
	}
	*x = payload
}

// entries returns the number of entries in the data.

name: gitStream workflow automation
run-name: |
  /:\ gitStream: PR #${{ fromJSON(fromJSON(github.event.inputs.client_payload)).pullRequestNumber }} from ${{ github.event.inputs.full_repository }}

on:
  workflow_dispatch:
    inputs:
      client_payload:
        description: The Client payload
        required: true
      full_repository:
        description: the repository name include the owner in `owner/repo_name` format

	// in theory we should close the fd, but it's just a test..

	assert.Equal(t, m.Payload.(*zdsapi.WorkloadRequest_Add).Add.Uid, uid)
	// send ack so the server doesn't wait for us.
	sendAck(ztunClient)

	// second message should be the snap sent message
	m, fds = readRequest(t, ztunClient)
	assert.Equal(t, len(fds), 0)

	sent := m.Payload.(*zdsapi.WorkloadRequest_SnapshotSent).SnapshotSent
	if sent == nil {

		return
	}

	// Route directly to output.
	m.respWait = out
}

// RequestStream will send a single payload request and stream back results.
// 'requests' can be nil, in which case only req is sent as input.
// It will however take less resources.
func (m *muxClient) RequestStream(h HandlerID, payload []byte, requests chan []byte, responses chan Response) (*Stream, error) {
	if m.init {

				Payload: &zdsapi.WorkloadRequest_Add{
					Add: &zdsapi.AddWorkload{
						Uid:          uid,
						WorkloadInfo: wl.Workload,
					},
				},
			}, &fd)
		} else {
			log.Infof("netns not available for local pod %s. sending keep to ztunnel", uid)
			resp, err = conn.sendMsgAndWaitForAck(&zdsapi.WorkloadRequest{
				Payload: &zdsapi.WorkloadRequest_Keep{

          distribution: temurin
          java-version: 11
      - id: determine-sys-prop-args
        uses: actions/github-script@v7
        with:
          script: |
            if (context.payload.pull_request && context.payload.pull_request.head.repo.fork) {
                core.setOutput('sys-prop-args', '-DagreePublicBuildScanTermOfService=yes -DcacheNode=us')
            } else {

		// Test case with "X-Amz-Content-Sha256" not set so we can skip.
		{"", "", "", "", true},

		// Test case - 3.
		// Test case with "X-Amz-Content-Sha256" header set to  "UNSIGNED-PAYLOAD"
		// When "X-Amz-Content-Sha256" header is set to  "UNSIGNED-PAYLOAD", validation of content sha256 has to be skipped.
		{"X-Amz-Content-Sha256", unsignedPayload, "X-Amz-Credential", "", true},

		// Test case - 4.