Add streaming publisher to pubsub.Topic. (#76)

api/pubsub.capnp

@@ -7,10 +7,11 @@ $Go.import("github.com/wetware/ww/internal/api/pubsub");
 
 
 interface Topic {
-    publish   @0 (msg :Data) -> ();
+    publish   @0 (msg :Data) -> stream;
     subscribe @1 (chan :Sender(Data), buf :UInt16 = 32) -> ();
     name      @2 () -> (name :Text);
 
+
     using Sender = import "channel.capnp".Sender;
 }
 

pkg/pubsub/topic.go

@@ -53,6 +53,19 @@ func (t Topic) Publish(ctx context.Context, b []byte) error {
 	return f.Err()
 }
 
+// NewStream provides an interface for publishing large volumes of data
+// through a flow-controlled channel.   This will override the existing
+// FlowLimiter.
+func (t Topic) NewStream(ctx context.Context) Stream {
+	topic := api.Topic(t)
+	topic.SetFlowLimiter(bbr.NewLimiter(clock.System))
+
+	return Stream{
+		ctx:    ctx,
+		stream: stream.New(topic.Publish),
+	}
+}
+
 // PublishAsync submits a message for broadcast over the topic.  Unlike
 // Publish, it returns a future.  This is useful when applications must
 // publish a large volume of messages, and callers do not wish to spawn
@@ -62,12 +75,6 @@ func (t Topic) PublishAsync(ctx context.Context, b []byte) (casm.Future, capnp.R
 	return casm.Future(f), release
 }
 
-func message(b []byte) func(api.Topic_publish_Params) error {
-	return func(ps api.Topic_publish_Params) error {
-		return ps.SetMsg(b)
-	}
-}
-
 // Subscribe to the topic.  Callers MUST call the provided ReleaseFunc
 // when finished with the subscription, or a resource leak will occur.
 func (t Topic) Subscribe(ctx context.Context) (Subscription, capnp.ReleaseFunc) {
@@ -97,6 +104,29 @@ func (t Topic) Subscribe(ctx context.Context) (Subscription, capnp.ReleaseFunc)
 		}
 }
 
+type Stream struct {
+	ctx    context.Context
+	stream *stream.Stream[api.Topic_publish_Params]
+}
+
+func (s Stream) Publish(msg []byte) (err error) {
+	if s.stream.Call(s.ctx, message(msg)); !s.stream.Open() {
+		err = s.Close()
+	}
+
+	return
+}
+
+func (s Stream) Close() error {
+	return s.stream.Wait()
+}
+
+func message(b []byte) func(api.Topic_publish_Params) error {
+	return func(ps api.Topic_publish_Params) error {
+		return ps.SetMsg(b)
+	}
+}
+
 /*
 	Topic Server
 */