bldr/resource/resource-rpc-data-stream.go

Rama master · réplica de lectura del repositorio original.

package resource

import (
	"errors"

	"github.com/aperturerobotics/starpc/rpcstream"
	"github.com/aperturerobotics/starpc/srpc"
)

// resourceRpcDataStream adapts negotiated Resource data to the SRPC codec.
type resourceRpcDataStream struct {
	// ResourceRpcStream carries the negotiated route and data packets.
	ResourceRpcStream
	// retain leaves closure to ExecCall on clients and the route reader on servers.
	retain bool
	// onSendError ends responses when a failed transport cannot carry nested
	// completion to the caller.
	onSendError func()
}

// CloseSend leaves unary route closure to ExecCall and forwards streaming closure.
func (s *resourceRpcDataStream) CloseSend() error {
	if s.retain {
		return nil
	}
	return s.ResourceRpcStream.CloseSend()
}

// Send forwards only data; the Resource route is already negotiated.
func (s *resourceRpcDataStream) Send(packet *rpcstream.RpcStreamPacket) error {
	// Require an SRPC data packet before forwarding it over the Resource route.
	data, ok := packet.GetBody().(*rpcstream.RpcStreamPacket_Data)
	if !ok {
		return rpcstream.ErrUnexpectedPacket
	}

	// A failed response transport cannot deliver nested completion to its
	// caller. An oversized packet leaves the transport intact, so the handler's
	// error completion still reaches the caller.
	err := s.ResourceRpcStream.Send(&ResourceRpcPacket{Body: &ResourceRpcPacket_Data{Data: data.Data}})
	if err != nil && !errors.Is(err, srpc.ErrMessageTooLarge) && s.onSendError != nil {
		s.onSendError()
	}
	return err
}

// Recv rejects a repeated handshake after the route has been acknowledged.
func (s *resourceRpcDataStream) Recv() (*rpcstream.RpcStreamPacket, error) {
	// Read the next packet and accept only data frames.
	packet, err := s.ResourceRpcStream.Recv()
	if err != nil {
		return nil, err
	}
	data, ok := packet.GetBody().(*ResourceRpcPacket_Data)
	if !ok {
		return nil, rpcstream.ErrUnexpectedPacket
	}
	return &rpcstream.RpcStreamPacket{Body: &rpcstream.RpcStreamPacket_Data{Data: data.Data}}, nil
}

// _ is a type assertion.
var _ rpcstream.RpcStream = (*resourceRpcDataStream)(nil)