Documentation
¶
Overview ¶
Package grpcbridge provides a gRPC gateway that bridges a DDS participant to gRPC clients using JSON encoding.
The DDSBridge service exposes three RPCs:
- Subscribe(SubscribeRequest) → stream(Sample): server-streaming, delivers DDS samples as they arrive.
- Publish(PublishRequest) → PublishAck: unary publish of a single sample.
- StreamPublish(stream PublishRequest) → PublishAck: client-streaming for high-throughput writes; returns the total count when the stream closes.
Messages are JSON-encoded over standard gRPC framing (Content-Type: application/grpc+json). Go clients set this via grpc.ForceCodec(grpcbridge.JSONCodec{}) on the dial options.
Usage:
p, _ := rtps.New(dds.Domain(0))
b := grpcbridge.New(p, grpcbridge.Options{})
lis, _ := net.Listen("tcp", ":9090")
b.Server().Serve(lis)
Index ¶
- func ApplyConfig(b *Bridge, cfg *Config) error
- func RegisterDDSBridgeServer(s *grpc.Server, srv DDSBridgeServer)
- type Bridge
- func (b *Bridge) Close() error
- func (b *Bridge) Publish(ctx context.Context, req *PublishRequest) (*PublishAck, error)
- func (b *Bridge) Server() *grpc.Server
- func (b *Bridge) StreamPublish(stream grpc.ClientStreamingServer[PublishRequest, PublishAck]) error
- func (b *Bridge) Subscribe(req *SubscribeRequest, stream grpc.ServerStreamingServer[Sample]) error
- type Config
- type DDSBridgeClient
- type DDSBridgeServer
- type FilterFunc
- type JSONCodec
- type Options
- type PublishAck
- type PublishRequest
- type Sample
- type SubscribeRequest
- type TopicConfig
- type TransformFunc
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ApplyConfig ¶
ApplyConfig pre-subscribes the topics listed in cfg on p. The Bridge opts are updated with the config's auth token when opts.AuthToken is empty.
func RegisterDDSBridgeServer ¶
func RegisterDDSBridgeServer(s *grpc.Server, srv DDSBridgeServer)
RegisterDDSBridgeServer registers srv on s.
Types ¶
type Bridge ¶
type Bridge struct {
// contains filtered or unexported fields
}
Bridge implements DDSBridgeServer and manages lazy subscribers/publishers.
func New ¶
func New(p dds.Participant, opts Options) *Bridge
New returns a Bridge wrapping p. Call Server().Serve(lis) to start accepting gRPC connections.
func (*Bridge) Publish ¶
func (b *Bridge) Publish(ctx context.Context, req *PublishRequest) (*PublishAck, error)
Publish implements DDSBridgeServer.
func (*Bridge) StreamPublish ¶
func (b *Bridge) StreamPublish(stream grpc.ClientStreamingServer[PublishRequest, PublishAck]) error
StreamPublish implements DDSBridgeServer.
func (*Bridge) Subscribe ¶
func (b *Bridge) Subscribe(req *SubscribeRequest, stream grpc.ServerStreamingServer[Sample]) error
Subscribe implements DDSBridgeServer.
type Config ¶
type Config struct {
// Listen is the gRPC listen address (e.g. ":9090"). Used by LoadAndApply.
Listen string `yaml:"listen"`
// AuthToken, if non-empty, requires every RPC to carry a matching Bearer token.
AuthToken string `yaml:"auth_token"`
// Topics lists topics to pre-subscribe when the bridge starts.
Topics []TopicConfig `yaml:"topics"`
}
Config is the bridge configuration loaded from a YAML file.
Example:
listen: ":9090"
auth_token: "shared-secret"
topics:
- name: "sensors/temperature"
qos: "reliable"
- name: "vehicle/speed"
qos: "best_effort"
func LoadConfig ¶
LoadConfig reads a YAML config file at path and returns the parsed Config.
type DDSBridgeClient ¶
type DDSBridgeClient interface {
Subscribe(ctx context.Context, req *SubscribeRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Sample], error)
Publish(ctx context.Context, req *PublishRequest, opts ...grpc.CallOption) (*PublishAck, error)
StreamPublish(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PublishRequest, PublishAck], error)
}
DDSBridgeClient is the client-side interface for DDSBridge.
func NewClient ¶
func NewClient(ctx context.Context, addr string, extra ...grpc.DialOption) (DDSBridgeClient, *grpc.ClientConn, error)
NewClient returns a DDSBridgeClient that speaks to addr using JSON encoding. The returned conn must be closed by the caller when done.
func NewRawClient ¶
func NewRawClient(cc grpc.ClientConnInterface) DDSBridgeClient
NewRawClient wraps an existing *grpc.ClientConn as a DDSBridgeClient. The caller is responsible for closing the connection.
type DDSBridgeServer ¶
type DDSBridgeServer interface {
Subscribe(*SubscribeRequest, grpc.ServerStreamingServer[Sample]) error
Publish(context.Context, *PublishRequest) (*PublishAck, error)
StreamPublish(grpc.ClientStreamingServer[PublishRequest, PublishAck]) error
}
DDSBridgeServer is implemented by Bridge and registered with a grpc.Server.
type FilterFunc ¶
FilterFunc decides whether to forward a sample. Return false to drop it.
type JSONCodec ¶
type JSONCodec struct{}
JSONCodec encodes gRPC messages as JSON. Register it via encoding.RegisterCodec(JSONCodec{}) or grpc.ForceCodec(JSONCodec{}).
type Options ¶
type Options struct {
// AuthToken, if non-empty, requires every RPC to carry the metadata key
// authorization: Bearer <AuthToken>
AuthToken string
// QoS is applied to all subscribers and publishers created by the bridge.
QoS dds.QoS
// Filter, if non-nil, is called for each outbound sample (Subscribe path).
// Return false to drop the sample.
Filter FilterFunc
// Transform, if non-nil, rewrites the payload of each outbound sample
// (Subscribe path) before delivery to the gRPC client.
Transform TransformFunc
}
Options configures a Bridge.
type PublishAck ¶
type PublishAck struct {
Count uint32 `json:"count"`
}
PublishAck is the response from Publish and StreamPublish RPCs.
type PublishRequest ¶
PublishRequest is the request message for Publish and StreamPublish RPCs.
type Sample ¶
type Sample struct {
Topic string `json:"topic"`
Payload []byte `json:"payload"`
SequenceNumber uint64 `json:"seq_num"`
TimestampNs int64 `json:"timestamp_ns"`
WriterGUID []byte `json:"writer_guid"`
}
Sample is a DDS sample delivered by the Subscribe RPC.
type SubscribeRequest ¶
type SubscribeRequest struct {
Topic string `json:"topic"`
}
SubscribeRequest is the request message for the Subscribe RPC.
type TopicConfig ¶
type TopicConfig struct {
// Name is the DDS topic name (e.g. "sensors/temperature").
Name string `yaml:"name"`
// QoS is "reliable" or "best_effort". Defaults to DefaultQoS if unset.
QoS string `yaml:"qos"`
}
TopicConfig configures a single topic in the bridge.
Source Files