Discussion: package pubsub

I'd like to start thinking about a design for package pubsub/msgbus/...

Important use cases:

An async transport for existing/typical RPC messaging
A foundation for event sourcing
A way to manage streaming
Others?

Would like support for:

Kafka
RabbitMQ
NATS
NSQ
AWS SQS
GCP PubSub
Weave Mesh
Others?

Prior art:

Open questions:

What are the core abstractions and interfaces?
Is streaming significantly different than event sourcing and transport use cases?
What is the lowest common denominator for serialization?
Others?

Shall we start a document for this or are you going to edit the above message to keep track of state?

RFC 7464 - JSON Text Sequences could be a good candidate for the JSON streaming transport. Some of the general streaming protocol concepts discussed here are relevant as well.

Streaming RPC makes me think of WebSockets and gRPC streaming. Go examples for streaming gRPC are here: http://www.grpc.io/docs/tutorials/basic/go.html

@basvanbeek

@basvanbeek

Shall we start a document for this or are you going to edit the above message to keep track of state?

I'd like to keep discussion in the issue for now. I'll update the OP as necessary.

—

@intellidynamics

I would avoid a “bus” concept (centralized pipe) but use independent synchronous / asynchronous connections between tasks or services.

For event sourcing (typical pub/sub) the logical domain language is absolutely in terms of abstract topics, publishers, and consumers. But I think the nature of the beast requires us to manage an actual bus, n.b. connections, state management, etc. Details varying based on implementation.

Otherwise thanks for the great feedback. I don't think we need to rule any of your ideas out at this stage.

—

@jmank88

RFC 7464 - JSON Text Sequences could be a good candidate for the JSON streaming transport.

Great, thanks for the link, I've added it to Prior Art (above) and will review. In general I'd expect details like JSON to be implemented by user-provided serialization modules; I wouldn't want to enforce anything at that level in Go kit's opinions. It's also reminded me of EventSourcing and server-sent-events, also added to Prior Art.

—

@ChrisHines

Streaming RPC makes me think of WebSockets and gRPC streaming.

Agreed; I think gRPC streaming is probably the benchmark here, and (at least at the moment) I'd expect any Go kit streaming API to very closely mirror the gRPC model. It's also very close to how Asim modeled things in https://github.com/micro/go-micro.

Starting with pubsub, as it's easiest and we have a lot of prior art. Here's my first draft of a set of interfaces, basically lifted verbatim from basvanbeek/pubsub via gizmo/pubsub.

// Publisher is a minimal interface for publishing messages to a pool of
// subscribers. Publishers are probably (but not necessarily) sending to a
// message bus.
//
// Most paramaterization of the publisher (topology restrictions like a topic,
// exchange, or specific message type; queue or buffer sizes; etc.) should be
// done in the concrete constructor.
type Publisher interface {
    // Publish a single message, described by an io.Reader, to the given key.
    Publish(key string, r io.Reader) error

    // Stop the publisher.
    Stop() error
}

// Subscriber is a minimal interface for subscribing to published messages.
// Subscribers are probably (but not necessarily) receiving from a message bus.
//
// Most paramaterization of the subscriber (topology restrictions like a topic,
// exchange, or specific message type; queue or buffer sizes; etc.) should be
// done in the concrete constructor.
type Subscriber interface {
    // Start returns a channel of messages that the caller should consume.
    // Failure to keep up with the incoming messages will have different
    // consequences depending on the concrete implementation of the subscriber.
    //
    // The channel will be closed when the subscriber encounters an error, or
    // when the caller invokes Stop, whichever comes first.
    Start() <-chan Message

    // Err returns the error that was responsible for closing the channel of
    // incoming messages.
    Err() error

    // Stop the subscriber, closing the channel that was returned by Start.
    Stop() error
}

// Message is a minimal interface to describe payloads received by subscribers.
// Clients may type-assert to more concrete types (e.g. pubsub/kafka.Message) to
// get access to more specific behaviors.
type Message interface {
    // Messages implement io.Reader to access the payload data.
    io.Reader

    // Done indicates the client is finished with the message, and the
    // underlying implementation may free its resources. Clients should ensure
    // to call Done for every received message.
    Done() error
}

Questions on my side:

Is there a reason Publish can't take an io.Reader?
Is there a reason Message can't simply implement io.Reader?
For which use case(s) does Publish need its own key parameter?
I can imagine a few ways to build an async (RPC) transport on top of this set of interfaces. Before I get too deep, does anyone know of any prior art?

Streaming interfaces to come...

I can only talk about Kafka as I haven't seen other Go API's

yes it could take an io.Reader. The most commonly used Go API for Kafka uses []byte for the message payload and it is what protobuf messages serialize to, so that's why I used it in my pubsub take.
Key is not a broker generic item. In Kafka it is used as partitioning key for partitioned topics. I would like to be able to set a partitioning key for each message but it might make no sense for other broker back-ends.
Async RPC and Prior Art... often built on top of JMS, AMQP, Stomp and involving some messageID, correlationID and a return path/topic.

I'm not sure about this current interface for Publisher. The way I've understood pubsub, the entity doing the publishing isn't necessarily sticking around after Publishing something. So I'm uncertain about the need to have a method called Stop on it, because what would it do? Stop everything you have published, stop future publishes? Is there some nuance with a potential implementation that may require this?

I'm going to try and mock this out with rabbitmq real quick to see how this might work. On the whole I'm having trouble visualizing the components that you've laid out, as you've laid them out. However, it's possible that I'm just missing something.

Ok, I've created an example, with some example programs, just to see if I could adhere to the interface that is laid out as-is. It turns out I can, and I can see the purpose of every piece so far.

If you'd like to see the implementation, for review or reference, you can find it here:
https://github.com/Ayiga/kit/tree/pubsub-rabbitmq/pubsub/rabbitmq

So I'm uncertain about the need to have a method called Stop on it, because what would it do?

Effectively every implementation will require a stateful connection to the broker. Stop will gracefully terminate that connection, including (potentially) any in-flight requests. But, you got it in your example. Nice! Thanks. That's some validation.

This could potentially be a good base for an implementation: https://github.com/juju/pubsub.

I am really new to Go but I have some 2 cents about the pub/sub topic in general. I'd like to mention about a question which should stay in mind:

Am I Publishing an Event or am I Sending a Command?

You can link pub/sub thinking with TCP Multicast/Unicast. Multicast as publishing event, Unicast as sending command which publisher knows the destination. Maybe this might have an impact on your designs at some time.

IMHO the very first need of pub/sub to be dependent on a Bus, is to persist the messages on a queue (when there's any IPC error or such) until either the publisher or subscriber service is up and running again with the pending messages.
The second need may be to escalate the management of 'retry/compensation policy per event' into the Bus infrastructure.

In my experience, it's always been useful to model a message as a payload plus metadata in the form of standard/free-form headers/properties. E.g a message id property allows the generic implementation of dupe detection, poison message handling etc. Here are some properties from Java/JMS : https://docs.oracle.com/cd/E19798-01/821-1841/bnces/index.html

I am using go-kit with NATS and it's very nice. It's not very well done so i doubt i can contribute much.

But i was wondering if this has moved on in terms of a pub/sub abstraction being adapted in Nat's ?

All modules communication north south fully. Modules subscribe to each other via NATS to do east / west sharing of events.
Clients compose bit of modules together to make a full client.
This is definitely not the most efficient as often the server does all consolidation so that the client has just a single connection but that would mean another layer of composition / decomposition and i just don't think it's worth it.
All clients hold all the the data as a cache and writes do through to nats.
Hence the client is very fast anyway.

This is a basic cqrs system. It would be impossible to build cleanly with go-kit. Really thankful for this project.

Also above Nat's is all the rpc stuff.
Below Nat's is everything else.
Security, dB, logic, validation.
I am not sure this is very Performant, but it's more future proof i feel be side it's all transport, binding independent down there.

any progress / update?

Really wish there was some headway on this. I'm trying to design micro-services that are both GRPC servers but also pub/sub participants, and this would seal the deal on using go-kit. Right now, my only real options are Gizmo, Micro or roll-my-own. :(

+1 for this.

Great discussions everyone. I want to add my two cents to the list.

I suggest we don't let the versatility of message queues get in the way. Before trying to abstract message queue, abstract their usage. For example go kit already has a beautiful abstraction of transport. It should be much easier to rollout kafka/RabbitMQ server/client in the transport package, which merely use message queues as an async transport, leaving out all the rest. (event sourcing, replay, streaming, etc.) The newcomer NATS package is a good direction to follow.

If someday we were to build an event broadcasting library in go kit, it would not be too late until then for us to seek broker interfaces for various message buses.

@Reasno

I thought I'd provide the current implementation we're using to abstract our different messaging transports with a managed pubsub go-kit-"like" package:

// Publisher is a minimal interface for publishing messages to a pool of
// subscribers. Publishers are probably (but not necessarily) sending to a
// message bus.
//
// Most paramaterization of the publisher (topology restrictions like a topic,
// exchange, or specific message type; queue or buffer sizes; etc.) should be
// done in the concrete constructor.
type Publisher interface {
    // Publish a single message, described by an io.Reader, to the given key.
    // 
    // CHANGE(chrisprijic): added context, so transport can utilize it.
    Publish(ctx context.Context, key string, r io.Reader) error

    // Stop the publisher.
    Stop() error
}

// NOTE: This code only contains the parts that have been changed or added.

// Message is a minimal interface to describe payloads received by subscribers.
// Clients may type-assert to more concrete types (e.g. pubsub/kafka.Message) to
// get access to more specific behaviors.
type Message interface {
    // CHANGE(chrisprijic): Context can implement something similar to *http.Request, 
    // or can be made public on it's own. I use the first since I like the immutability
    // 
    //     Context() context.Context
    //     WithContext(context.Context) Message
    //
    // or
    //
    //     Context context.Context
    Context() context.Context
    WithContext(context.Context) Message

    // Messages implement io.Reader to access the payload data.
    io.Reader

    // Done indicates the client is finished with the message, and the
    // underlying implementation may free its resources. Clients should ensure
    // to call Done for every received message.
    // 
    // CHANGE(chrisprijic): ack was added so that we can specify cleanup AND 
    // whether or not we want to ack (in cases where we can). Ignore if not needed
    Done(ack bool) error
}

// NewManagedSubscriber can be used to wrap a subscriber into a managed
// transport.
//
// The function will subscribe to the given subscriber, use the decoder
// to decode the message, and send it to the given endpoint.
// 
// Message.Done() could be called on success of the endpoint running.
//
// This does bring up how to interpret the response/error from the endpoint...
// This also brings up how to interpret Message.Done(), meaning we're making 
// assumptions, should that be the case?
//
// -> I think this needs more discussion... 
func NewManagedSubscriber(
    e endpoint.Endpoint,
    subscriber Subscriber,
    decoder DecodeMessageFunc,
    options ...SubscriberOptions,
) *ManagedSubscriber {
    // implementation here
}

// Serve begins listening for messages and passing them to the endpoint:
func (s *ManagedSubscriber) Serve() error {
    // implementation here

    // something like the following:
    msgs := s.subscriber.Start()
    for msg := range msgs {
        // decode and send to endpoint, call msg.Done() when 
    }

    // check and return error if there were any
    return s.subscriber.Err()
}

// Stop can gracefully stop the subscriber:
func (s *ManagedSubscriber) Stop() error {
    return s.subscriber.Stop()
}

// NewManagedPublisher can be used to wrap a publisher into a managed
// transport.
// 
// The function returns an object that will publish messages provided to it 
// using the Publisher, to the given key. It can encode the message using the 
// EncodeMessageFunc.
//
// The returned transport can provide an endpoint with it's Endpoint() call, 
// which can be used like any other endpoint in go-kit.
func NewManagedPublisher(
    key string,
    publisher Publisher, 
    enc EncodeMessageFunc,
    options ...PublisherOptions,
) *ManagedPublisher {
    // implementation here
}

// Endpoint returns a useable endpoint for publishing messages on the publisher
// transport.
func (p *ManagedPublisher) Endpoint() endpoint.Endpoint {
    // implementation here
}

I think @Reasno makes a good point; however since all of our implementations had endpoint creation and the subscription message-loop looking refactor-able to the level above, we did so.

Since it's our implementation, we could make the assumptions posed by this. How these assumptions fit with go-kit, is another discussion...the "managed" code is secondary to the goal in this issue anyways. I think adding context availability throughout the package is the important part here.

Either way, I'm happy to help if I can. We've got our working implementation so hopefully the above can help others that might want to jump into it before go-kit has it.

Cross-referencing go-cloud's pubsub package as it has an abstraction and several implementations. Perhaps we can wrap an make those go-kit compatible endpoints?

@glerchundi

Interesting repo @glerchundi. I'm actually just started today experimenting a go-kit compatible implementation of google pubsub at https://github.com/revas/kit. 😉

Is something like this what you're looking to implement? It's far from complete, but Kafka is fleshed out and RabbitMQ is in progress. Starting to use this and would be nice to have it added to the go-kit standard lib, seems to be a popular request -

https://github.com/stratg5/pubsub

We already have amqp, and patterns should be possible to generalize from that. Closing this issue, if we want kafka I'm open to it.

@glerchundi

Interesting repo @glerchundi. I'm actually just started today experimenting a go-kit compatible implementation of google pubsub at https://github.com/revas/kit. 😉

Was watching this for 2 weeks before the close comment. Are there any plans to merge this into the mainline go-kit repository?

@zhongliangong

@zhongliangong yes soon. We are about to first test this implementation in our production before merging.

So idea of supporting multiple other providers like GCP PubSub was dropped? Or maybe there is somewhere plugin to support that?

Navigation Menu

Repository navigation

Discussion: package pubsub #295

9 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Discussion: package pubsub #295

Description

Activity

basvanbeek commented on Jun 15, 2016

jmank88 commented on Jun 15, 2016

ChrisHines commented on Jun 15, 2016

peterbourgon commented on Jun 15, 2016

peterbourgon commented on Jun 18, 2016

basvanbeek commented on Jun 18, 2016

Ayiga commented on Jul 4, 2016

Ayiga commented on Jul 4, 2016

peterbourgon commented on Jul 4, 2016

waigani commented on Aug 2, 2016

muratzorer commented on Aug 11, 2016

epsstan commented on Aug 15, 2016

joeblew99 commented on Apr 24, 2017

9 remaining items

welemon commented on Apr 12, 2018

jasonmacdonald commented on Apr 17, 2018

abdullah2993 commented on Jun 21, 2018

Reasno commented on Jun 24, 2018

chrisprijic commented on Jun 29, 2018

glerchundi commented on Mar 12, 2019

banh-gao commented on Mar 12, 2019

stratg5 commented on Jul 2, 2019

peterbourgon commented on Sep 1, 2019

zhongliangong commented on Sep 5, 2019

banh-gao commented on Sep 5, 2019

Mistic92 commented on Jun 3, 2022

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions