doc(broadcast): added documentation for broadcast options

Author: aleksander-vedvik

broadcast.go

@@ -46,7 +46,7 @@ func newBroadcastServer(serverOpts *serverOptions) *broadcastServer {
 		logger:    serverOpts.logger,
 		machineID: serverOpts.machineID,
 	}
-	srv.manager = broadcast.NewBroadcastManager(serverOpts.logger, createClient, srv.canceler, serverOpts.executionOrder, serverOpts.clientDialTimeout, serverOpts.reqTTL, serverOpts.shardBuffer, serverOpts.sendBuffer)
+	srv.manager = broadcast.NewBroadcastManager(serverOpts.logger, createClient, srv.canceler, serverOpts.executionOrder, serverOpts.clientDialTimeout, serverOpts.reqTTL, serverOpts.shardBuffer, serverOpts.sendBuffer, serverOpts.grpcDialOpts...)
 	srv.manager.AddAddr(srv.id, serverOpts.listenAddr, srv.machineID)
 	return srv
 }
@@ -112,30 +112,41 @@ func NewBroadcastOrchestrator(srv *Server) *BroadcastOrchestrator {
 
 type BroadcastOption func(*broadcast.BroadcastOptions)
 
+// WithSubset enables broadcasting to a subset of the servers in the view.
+// It has the same function as broadcast.To().
 func WithSubset(srvAddrs ...string) BroadcastOption {
 	return func(b *broadcast.BroadcastOptions) {
 		b.ServerAddresses = srvAddrs
 	}
 }
 
+// WithoutSelf prevents the server from broadcasting to itself.
 func WithoutSelf() BroadcastOption {
 	return func(b *broadcast.BroadcastOptions) {
 		b.SkipSelf = true
 	}
 }
 
+// ProgressTo allows the server to accept messages to the given method.
+// Should only be used if the ServerOption WithOrder() is used.
 func ProgressTo(method string) BroadcastOption {
 	return func(b *broadcast.BroadcastOptions) {
 		b.ProgressTo = method
 	}
 }
 
+// AllowDuplication allows the server to broadcast more than once
+// to the same RPC method for a particular broadcast request.
 func AllowDuplication() BroadcastOption {
 	return func(b *broadcast.BroadcastOptions) {
 		b.AllowDuplication = true
 	}
 }
 
+// WithRelationToRequest allows for broadcasting outside a
+// server handler related to a specific broadcastID.
+// It is not recommended to use this method. Use the broadcast
+// struct provided with a broadcast request instead.
 func WithRelationToRequest(broadcastID uint64) BroadcastOption {
 	return func(b *broadcast.BroadcastOptions) {
 		b.RelatedToReq = broadcastID

broadcast/manager.go

@@ -28,8 +28,8 @@ type manager struct {
 	logger *slog.Logger
 }
 
-func NewBroadcastManager(logger *slog.Logger, createClient func(addr string, dialOpts []grpc.DialOption) (*Client, error), canceler func(broadcastID uint64, srvAddrs []string), order map[string]int, dialTimeout, reqTTL time.Duration, shardBuffer, sendBuffer int) Manager {
-	router := NewRouter(logger, createClient, canceler, dialTimeout)
+func NewBroadcastManager(logger *slog.Logger, createClient func(addr string, dialOpts []grpc.DialOption) (*Client, error), canceler func(broadcastID uint64, srvAddrs []string), order map[string]int, dialTimeout, reqTTL time.Duration, shardBuffer, sendBuffer int, dialOpts ...grpc.DialOption) Manager {
+	router := NewRouter(logger, createClient, canceler, dialTimeout, dialOpts...)
 	state := NewState(logger, router, order, reqTTL, shardBuffer, sendBuffer)
 	router.registerState(state)
 	return &manager{

clientserver.go

@@ -306,7 +306,6 @@ func (srv *ClientServer) verify(req *Message) error {
 
 func createClient(addr string, dialOpts []grpc.DialOption) (*broadcast.Client, error) {
 	// necessary to ensure correct marshalling and unmarshalling of gorums messages
-	// TODO: find a better solution
 	dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.CallContentSubtype(ContentSubtype)))
 	opts := newManagerOptions()
 	opts.grpcDialOpts = dialOpts

server.go

@@ -180,9 +180,10 @@ type serverOptions struct {
 	// running in a docker container it is useful to listen to the loopback
 	// address and use forwarding from the host. If not this option is not given,
 	// the listen address used on the gRPC listener will be used instead.
-	listenAddr string
-	allowList  map[string]string
-	auth       *authentication.EllipticCurve
+	listenAddr   string
+	allowList    map[string]string
+	auth         *authentication.EllipticCurve
+	grpcDialOpts []grpc.DialOption
 }
 
 // ServerOption is used to change settings for the GorumsServer
@@ -244,6 +245,15 @@ func WithClientDialTimeout(dialTimeout time.Duration) ServerOption {
 	}
 }
 
+// WithServerGrpcDialOptions returns a ServerOption which sets any gRPC dial options
+// the Broadcast Router should use when connecting to each client.
+func WithServerGrpcDialOptions(opts ...grpc.DialOption) ServerOption {
+	return func(o *serverOptions) {
+		o.grpcDialOpts = make([]grpc.DialOption, 0, len(opts))
+		o.grpcDialOpts = append(o.grpcDialOpts, opts...)
+	}
+}
+
 // WithShardBuffer returns a ServerOption which sets the buffer size
 // of the shards. A higher shard size uses more memory but saves time
 // when the number of broadcast requests is large.