grpc

Documentation

Overview

Package grpc provides grpc utilities.

Index

• Variables
• func Dial(ctx context.Context, address string, logger logging.Logger, ...) (rpc.ClientConn, error)
• func EnsureTimeoutUnaryClientInterceptor(ctx context.Context, method string, req, reply interface{}, ...) error
• func EnsureTimeoutUnaryServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, ...) (interface{}, error)
• func InferSignalingServerAddress(address string) (string, bool, bool)
• type ForeignResource
  • func NewForeignResource(name resource.Name, conn rpc.ClientConn) *ForeignResource
  • func (res *ForeignResource) NewStub() grpcdynamic.Stub
  • func (res *ForeignResource) Reconfigure(ctx context.Context, deps resource.Dependencies, conf resource.Config) error
• type OnTrackCB
• type ReconfigurableClientConn
  • func (c *ReconfigurableClientConn) Close() error
  • func (c *ReconfigurableClientConn) Invoke(ctx context.Context, method string, args, reply interface{}, ...) error
  • func (c *ReconfigurableClientConn) NewStream(ctx context.Context, desc *googlegrpc.StreamDesc, method string, ...) (googlegrpc.ClientStream, error)
  • func (c *ReconfigurableClientConn) PeerConn() *webrtc.PeerConnection
  • func (c *ReconfigurableClientConn) ReplaceConn(conn rpc.ClientConn)
• type SharedConn
  • func (sc *SharedConn) AddOnTrackSub(name resource.Name, onTrackCB OnTrackCB)
  • func (sc *SharedConn) Close() error
  • func (sc *SharedConn) GenerateEncodedOffer() (string, error)
  • func (sc *SharedConn) GrpcConn() *ReconfigurableClientConn
  • func (sc *SharedConn) Invoke(ctx context.Context, method string, args, reply interface{}, ...) error
  • func (sc *SharedConn) NewStream(ctx context.Context, desc *googlegrpc.StreamDesc, method string, ...) (googlegrpc.ClientStream, error)
  • func (sc *SharedConn) PeerConn() *webrtc.PeerConnection
  • func (sc *SharedConn) ProcessEncodedAnswer(encodedAnswer string) error
  • func (sc *SharedConn) RemoveOnTrackSub(name resource.Name)
  • func (sc *SharedConn) ResetConn(conn rpc.ClientConn, moduleLogger logging.Logger)

Variables

var DefaultMethodTimeout = 10 * time.Minute

DefaultMethodTimeout is the default context timeout for all inbound gRPC methods and all outbound gRPC methods to modules, only used when no deadline is set on the context.

var DefaultWebRTCConfiguration = webrtc.Configuration{
	ICEServers: []webrtc.ICEServer{
		{
			URLs: []string{"stun:global.stun.twilio.com:3478"},
		},
	},
}

DefaultWebRTCConfiguration is the default configuration to use.

var UnimplementedError = status.Error(codes.Unimplemented, codes.Unimplemented.String())

UnimplementedError is returned any time a gRPC method is unimplemented.

Functions

func Dial

func Dial(ctx context.Context, address string, logger logging.Logger, opts ...rpc.DialOption) (rpc.ClientConn, error)

Dial dials a gRPC server. `ctx` can be used to set a timeout/deadline for Dial. However, the signaling server may have other timeouts which may prevent the full timeout from being respected.

func EnsureTimeoutUnaryClientInterceptor

func EnsureTimeoutUnaryClientInterceptor(
	ctx context.Context,
	method string, req, reply interface{},
	cc *grpc.ClientConn,
	invoker grpc.UnaryInvoker,
	opts ...grpc.CallOption,
) error

EnsureTimeoutUnaryClientInterceptor sets a default timeout on the context if one is not already set. To be called as the first unary client interceptor.

func EnsureTimeoutUnaryServerInterceptor

func EnsureTimeoutUnaryServerInterceptor(ctx context.Context, req interface{},
	info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (interface{}, error)

EnsureTimeoutUnaryServerInterceptor sets a default timeout on the context if one is not already set. To be called as the first unary server interceptor.

func InferSignalingServerAddress

func InferSignalingServerAddress(address string) (string, bool, bool)

InferSignalingServerAddress returns the appropriate WebRTC signaling server address if it can be detected. Returns the address, if the endpoint is secure, and if found. TODO(RSDK-235): remove hard coding of signaling server address and prefer SRV lookup instead.

Types

type ForeignResource

type ForeignResource struct {
	resource.Named
	resource.TriviallyCloseable
	// contains filtered or unexported fields
}

An ForeignResource is used to dynamically invoke RPC calls to resources that have their RPC information dervied on demand.

func NewForeignResource

func NewForeignResource(name resource.Name, conn rpc.ClientConn) *ForeignResource

NewForeignResource returns an ForeignResource for the given resource name and connection serving it.

func (*ForeignResource) NewStub

func (res *ForeignResource) NewStub() grpcdynamic.Stub

NewStub returns a new gRPC client stub used to communicate with the resource.

func (*ForeignResource) Reconfigure

func (res *ForeignResource) Reconfigure(ctx context.Context, deps resource.Dependencies, conf resource.Config) error

Reconfigure always fails.

type OnTrackCB

type OnTrackCB func(tr *webrtc.TrackRemote, r *webrtc.RTPReceiver)

OnTrackCB is the signature of the OnTrack callback function a resource may register with a SharedConn which supports WebRTC.

type ReconfigurableClientConn

type ReconfigurableClientConn struct {
	// contains filtered or unexported fields
}

ReconfigurableClientConn allows for the underlying client connections to be swapped under the hood.

func (*ReconfigurableClientConn) Close

func (c *ReconfigurableClientConn) Close() error

Close attempts to close the underlying client connection if there is one.

func (*ReconfigurableClientConn) Invoke

func (c *ReconfigurableClientConn) Invoke(
	ctx context.Context,
	method string,
	args, reply interface{},
	opts ...googlegrpc.CallOption,
) error

Invoke invokes using the underlying client connection. In the case of c.conn being closed in the middle of an Invoke call, it is expected that c.conn can handle that and return a well-formed error.

func (*ReconfigurableClientConn) NewStream

func (c *ReconfigurableClientConn) NewStream(
	ctx context.Context,
	desc *googlegrpc.StreamDesc,
	method string,
	opts ...googlegrpc.CallOption,
) (googlegrpc.ClientStream, error)

NewStream creates a new stream using the underlying client connection. In the case of c.conn being closed in the middle of a NewStream call, it is expected that c.conn can handle that and return a well-formed error.

func (*ReconfigurableClientConn) PeerConn

func (c *ReconfigurableClientConn) PeerConn() *webrtc.PeerConnection

PeerConn returns the backing PeerConnection object, if applicable. Nil otherwise.

func (*ReconfigurableClientConn) ReplaceConn

func (c *ReconfigurableClientConn) ReplaceConn(conn rpc.ClientConn)

ReplaceConn replaces the underlying client connection with the connection passed in. This does not close the old connection, the caller is expected to close it if needed.

type SharedConn

type SharedConn struct {
	// contains filtered or unexported fields
}

SharedConn wraps both a GRPC connection & (optionally) a peer connection & controls access to both. For modules, the grpc connection is over a Unix socket. The WebRTC `PeerConnection` is made separately. The `SharedConn` continues to implement the `rpc.ClientConn` interface by pairing up the two underlying connections a client may want to communicate over.

Users of SharedConn must serialize calls to `ResetConn`, `GenerateEncodedOffer`, `ProcessEncodedAnswer` and `Close`.

The lifetime of SharedConn objects are a bit nuanced. SharedConn objects are expected to be handed to resource Client objects. When connections break and become reestablished, those client objects are not recreated, but rather we swap in new connection objects under the hood.

This is because resource management controls updating the graph nodes with Client objects only when calls to `Reconfigure` are made. But connections can break and heal without any calls to Reconfigure.

Thus the lifetime of a SharedConn for a client is _not_: 1) Object initialize 2) Working 3) Close 4) Object destruction And back to initialization.

But rather: 1) Object initialization 2) `ResetConn` 3) `Close` 4) `ResetConn` ... 5) Object destruction (at shutdown or resource removed from config).

Each call to `ResetConn` is a new "generation" of connections. When a new gRPC connection is handed to a SharedConn, the SharedConn will start the process of establishing a new PeerConnection. A call to access the PeerConnection will block until this process completes, either successfully or not.

When a PeerConnection cannot be established, the SharedConn is in a degraded state where the gRPC connection is still valid. The caller has the option of either disabling functionality intended for the PeerConnection, or implementing that functionality over gRPC (which would presumably be less performant).

func (*SharedConn) AddOnTrackSub

func (sc *SharedConn) AddOnTrackSub(name resource.Name, onTrackCB OnTrackCB)

AddOnTrackSub adds an OnTrack subscription for the resource.

func (*SharedConn) Close

func (sc *SharedConn) Close() error

Close closes a shared connection.

func (*SharedConn) GenerateEncodedOffer

func (sc *SharedConn) GenerateEncodedOffer() (string, error)

GenerateEncodedOffer creates a WebRTC offer that's JSON + base64 encoded. If an error is returned, `SharedConn.PeerConn` will return nil until a following `Reset`.

func (*SharedConn) GrpcConn

func (sc *SharedConn) GrpcConn() *ReconfigurableClientConn

GrpcConn returns a gRPC capable client connection.

func (*SharedConn) Invoke

func (sc *SharedConn) Invoke(
	ctx context.Context,
	method string,
	args, reply interface{},
	opts ...googlegrpc.CallOption,
) error

Invoke forwards to the underlying GRPC Connection.

func (*SharedConn) NewStream

func (sc *SharedConn) NewStream(
	ctx context.Context,
	desc *googlegrpc.StreamDesc,
	method string,
	opts ...googlegrpc.CallOption,
) (googlegrpc.ClientStream, error)

NewStream forwards to the underlying GRPC Connection.

func (*SharedConn) PeerConn

func (sc *SharedConn) PeerConn() *webrtc.PeerConnection

PeerConn returns a WebRTC PeerConnection capable of sending video/audio data. A call to PeerConn concurrent with `ResetConn` or `Close` is allowed to return either the old or new connection.

func (*SharedConn) ProcessEncodedAnswer

func (sc *SharedConn) ProcessEncodedAnswer(encodedAnswer string) error

ProcessEncodedAnswer sets the remote description to the answer and waits for the connection to be ready as per the negotiation channel being opened. The answer is expected to be JSON + base64 encoded. If an error is returned, `SharedConn.PeerConn` will return nil until a following `Reset`.

func (*SharedConn) RemoveOnTrackSub

func (sc *SharedConn) RemoveOnTrackSub(name resource.Name)

RemoveOnTrackSub removes an OnTrack subscription for the resource.

func (*SharedConn) ResetConn

func (sc *SharedConn) ResetConn(conn rpc.ClientConn, moduleLogger logging.Logger)

ResetConn acts as a constructor for `SharedConn`. ResetConn replaces the underlying connection objects in addition to some other initialization.

The first call to `ResetConn` is guaranteed to happen before any access to connection objects happens. But subequent calls can be entirely asynchronous to components/services accessing `SharedConn` for connection objects.