I have an API service which needs to consume a Kafka compacted topic before to be considered ready to handle any request traffic. How can I determine its readiness?

In this case, it would be best to know how many messages are still left to process before to have catched up with the latest messages on the topic. It seems there is no way to know this information currently.

I suggest the partitions offset to be retrieved at the time the consumer subscribes to a topic (from all partitions consumed). An alternative would be to include a field to messages, a flag IsLatest for example to let the consumer knows that a message was the last one at the time of retrieval (of course the offsets are keeping growing therefore it should not be considered as an absolute indication, it's time sensitive).

The best I can do as a workaround for now is to infer it. I can use the throughput and assume I've reached the end of the topic once the number of message/sec significantly dropped (during catch up phase, the consumer processes as much messages as it can, then after only as much as "live" events currently produced on the topic).

I'll be happy to know other alternatives people might have come up with. Thanks.

Hello to all contributors!

I am using watermill in my VoIP tech stack to make thousands calls per day and it works perfectly. Thanks to all who made this awesome project!

The NATS documentation says the following:

So my question is simple - Any plans to add support of JetStream instead of NATS Streaming?

When passing a nil DB the code that checks the db == nill is false because the beginner is an interface so the nil DB satisfies the interface but is nil. See this link for reference. https://glucn.medium.com/golang-an-interface-holding-a-nil-value-is-not-nil-bb151f472cc7

func NewSubscriber(db beginner, config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error) {
	if db == nil { // <- this is always false even if we sent a nil db
		return nil, errors.New("db is nil")
	}

So in other words there is an assumption that we are caching the case of receiving a nil value and returning an error but db == nill return false in all cases.

The next page explores some ways to check that an interface is nil: https://mangatmodi.medium.com/go-check-nil-interface-the-right-way-d142776edef1

The chi.Router is not used and I believe it is not intentional. I have added tests to cover the expected behaviour. go fmt has added a small diff in pubsub.go.

Also out of curiosity, is there any benefit to do

	wait := make(chan struct{})
	go func() {
		<-wait
		server.Close()
	}()

	return func() { close(wait) }

instead of directly?

	return func() { server.Close() }

Thanks

Following fails on my Windows. It looks like watermill is having some exotic external dependencies which do not work out of the box on Windows, neither it is documented bzr is required.

This makes it hard to get a working environment on at least Windows to use watermill. Would be great if this could be simplyfied. Or maybe some of external deps to be internalized.

$ go get -u github.com/ThreeDotsLabs/watermill                                                                                        go: finding github.com/armon/go-metrics latest                                                                                        go: finding github.com/streadway/amqp latest                                                                                          go: finding github.com/pascaldekloe/goe latest                                                                                        go: finding golang.org/x/net latest
go: finding golang.org/x/sync latest
go: finding golang.org/x/oauth2 latest
go: finding github.com/eapache/go-xerial-snappy latest
go: finding golang.org/x/crypto latest
go: finding github.com/google/btree latest
go: finding golang.org/x/sys latest
go: finding golang.org/x/lint latest
go: finding google.golang.org/genproto latest
go: finding golang.org/x/build latest
go: finding github.com/shurcooL/gopherjslib latest
go: labix.org/v2/[email protected]: bzr branch --use-existing-dir https://launchpad.net/mgo/v2 . in C:\private-s tuff\go-workspace\pkg\mod\cache\vcs\ca61c737a32b1e09a0919e15375f9c2b6aa09860cc097f1333b3c3d29e040ea8: exit status 4:
    bzr: ERROR: httplib.IncompleteRead: IncompleteRead(34 bytes read)

    Traceback (most recent call last):
         File "bzrlib\commands.pyo", line 920, in exception_to_return_code
         File "bzrlib\commands.pyo", line 1131, in run_bzr
         File "bzrlib\commands.pyo", line 673, in run_argv_aliases
         File "bzrlib\commands.pyo", line 695, in run
         File "bzrlib\cleanup.pyo", line 136, in run_simple
         File "bzrlib\cleanup.pyo", line 166, in _do_with_cleanups
         File "bzrlib\builtins.pyo", line 1438, in run
         File "bzrlib\controldir.pyo", line 779, in open_tree_or_branch
         File "bzrlib\controldir.pyo", line 459, in _get_tree_branch
         File "bzrlib\bzrdir.pyo", line 1082, in open_branch
         File "bzrlib\branch.pyo", line 2375, in open
         File "bzrlib\controldir.pyo", line 687, in open
         File "bzrlib\controldir.pyo", line 716, in open_from_transport
         File "bzrlib\transport\__init__.pyo", line 1718, in do_catching_redirections
         File "bzrlib\controldir.pyo", line 704, in find_format
         File "bzrlib\controldir.pyo", line 1149, in find_format
         File "C:/Program Files (x86)/Bazaar/plugins\git\__init__.py", line 235, in probe_transport
         File "C:/Program Files (x86)/Bazaar/plugins\git\__init__.py", line 182, in probe_http_transport
         File "socket.pyo", line 348, in read
         File "httplib.pyo", line 522, in read
         File "httplib.pyo", line 565, in _read_chunked
    IncompleteRead: IncompleteRead(34 bytes read)

    bzr 2.5.1 on python 2.6.6 (Windows-post2008Server-6.2.9200)
    arguments: ['bzr', 'branch', '--use-existing-dir',
        'https://launchpad.net/mgo/v2', '.']
    plugins: bzrtools[2.5.0], changelog_merge[2.5.1], colo[0.4.0],
        explorer[1.2.2], fastimport[0.14.0dev], git[0.6.8], launchpad[2.5.1],
        loom[2.3.0dev], netrc_credential_store[2.5.1], news_merge[2.5.1],
        pipeline[1.4.0], qbzr[0.22.3], rewrite[0.6.4dev], svn[1.2.2],
        upload[1.2.0dev], xmloutput[0.8.8]
    encoding: 'cp1252', fsenc: 'mbcs', lang: 'en_US.UTF-8'

    *** Bazaar has encountered an internal error.  This probably indicates a
          bug in Bazaar.  You can help us fix it by filing a bug report at
              https://bugs.launchpad.net/bzr/+filebug
          including this traceback and a description of the problem.
go: finding grpc.go4.org latest
go: finding golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e
go: golang.org/x/[email protected]: unknown revision 1e06a53dbb7e
go: finding github.com/shurcooL/github_flavored_markdown latest
go: finding github.com/shurcooL/httpgzip latest
go: launchpad.net/[email protected]: bzr branch --use-existing-dir https://launchpad.net/~niemeyer/gocheck/t runk . in C:\private-stuff\go-workspace\pkg\mod\cache\vcs\f46ce2ae80d31f9b0a29099baa203e3b6d269dace4e5357a2cf74bd109e13339: exit stat us 4:
    bzr: ERROR: httplib.IncompleteRead: IncompleteRead(34 bytes read)

    Traceback (most recent call last):
         File "bzrlib\commands.pyo", line 920, in exception_to_return_code
         File "bzrlib\commands.pyo", line 1131, in run_bzr
         File "bzrlib\commands.pyo", line 673, in run_argv_aliases
         File "bzrlib\commands.pyo", line 695, in run
         File "bzrlib\cleanup.pyo", line 136, in run_simple
         File "bzrlib\cleanup.pyo", line 166, in _do_with_cleanups
         File "bzrlib\builtins.pyo", line 1438, in run
         File "bzrlib\controldir.pyo", line 779, in open_tree_or_branch
         File "bzrlib\controldir.pyo", line 459, in _get_tree_branch
         File "bzrlib\bzrdir.pyo", line 1082, in open_branch
         File "bzrlib\branch.pyo", line 2375, in open
         File "bzrlib\controldir.pyo", line 687, in open
         File "bzrlib\controldir.pyo", line 716, in open_from_transport
         File "bzrlib\transport\__init__.pyo", line 1718, in do_catching_redirections
         File "bzrlib\controldir.pyo", line 704, in find_format
         File "bzrlib\controldir.pyo", line 1149, in find_format
         File "C:/Program Files (x86)/Bazaar/plugins\git\__init__.py", line 235, in probe_transport
         File "C:/Program Files (x86)/Bazaar/plugins\git\__init__.py", line 182, in probe_http_transport
         File "socket.pyo", line 348, in read
         File "httplib.pyo", line 522, in read
         File "httplib.pyo", line 565, in _read_chunked
     IncompleteRead: IncompleteRead(34 bytes read)

     bzr 2.5.1 on python 2.6.6 (Windows-post2008Server-6.2.9200)
     arguments: ['bzr', 'branch', '--use-existing-dir',
        'https://launchpad.net/~niemeyer/gocheck/trunk', '.']
    plugins: bzrtools[2.5.0], changelog_merge[2.5.1], colo[0.4.0],
        explorer[1.2.2], fastimport[0.14.0dev], git[0.6.8], launchpad[2.5.1],
        loom[2.3.0dev], netrc_credential_store[2.5.1], news_merge[2.5.1],
        pipeline[1.4.0], qbzr[0.22.3], rewrite[0.6.4dev], svn[1.2.2],
        upload[1.2.0dev], xmloutput[0.8.8]
    encoding: 'cp1252', fsenc: 'mbcs', lang: 'en_US.UTF-8'

    *** Bazaar has encountered an internal error.  This probably indicates a
          bug in Bazaar.  You can help us fix it by filing a bug report at
          https://bugs.launchpad.net/bzr/+filebug
          including this traceback and a description of the problem.
go get: error loading module requirements

Currently, there are 3 types of data provided in the kafkaMsg context which are partition number, offset and timestamp of the message. In my case, I will need kafkaMsg's key field to filter out some messages.

Would it be better if some method like MessageKeyFromCtx and setMessageKeyToCtx could be provided to use?

Thanks for your time and best regards.

I was looking for an SMTP-based MQ pub/sub capability: basically using gmail or any other SMTP provider as a "poor man's" MQ for small number of messages. Has anyone seen such a system that can be wrapped as a Watermill component?

The problem

We use the routing key in bindings.

The routing key is dynamic and in our example is an event name. Because the routing key is static, we don't have the flexibility we need.

In this PR, I change the variable to a function similar to routing keys in PublishConfig etc

Issue: Consumers happen to get stuck in a rebalance state due to call group.Close() improperly

Module: "github.com/ThreeDotsLabs/watermill-kafka/v2/pkg/kafka" v2.2.0

go version: 1.14.4

What did you do? start more than 2 consumers in a consumer group to consume a topic with 64 partitions.

What did you expect to see? each consumer will consume half partitions data properly.

What did you see instead? Each consumer repeatedly restarts without consuming messages, and this case can be reproduced steadily. detail log as follows: log.txt

Possible cause

In https://github.com/ThreeDotsLabs/watermill-kafka/blob/master/pkg/kafka/subscriber.go#L292, group.Consume() returns with no err due to consumer member change,

1. group.Close() should not be called since there is no error and the current member still wants to process some partitions' s messages.

2. instead, should continue call group.Consume() to get the new topic partitions claims and consume messages afterward.

3. Besides, according to the official example of the sarama library, https://github.com/Shopify/sarama/blob/master/examples/consumergroup/main.go#L102, group.Consume() should be called in an infinite loop.

Could you please help check the issue, thanks and best regards!

I'm setting up a non-Go system to pass CQRS commands to Watermill over amqp. In my first attempt trying to reproduce the message structure that Watermill expects, I missed the MessageUUIDHeaderKey header (_watermill_message_uuid) which resulted in an infinite Nack loop. Watermill rejects the message before reaching router middleware or the handler, so there's no opportunity to handle the malformed message (eg, w/ a dead letter queue) or otherwise break the infinite loop.

subscriber.go:218: level=ERROR msg="Processing message failed, sending nack" amqp_exchange_name=dev_primary amqp_queue_name=tasks.SomeTask err="missing _watermill_message_uuid header" topic=tasks.SomeTask

I suspect an error encountered during unmarshalling might also result in the same behavior.

Hey gang, I am opening this as a quick sanity check, I am using CQRS facade and ran into an interesting issue:

If a handler returns an error (or panics) it will retry that handler until the message is successfully processed. I tried using Retry middleware, but it does not work. The reason being, this line https://github.com/ThreeDotsLabs/watermill/blob/8840d3f3fae33785dc9a2ecdae196b83bdcaf01e/message/router/middleware/retry.go#L94 returns an err (which is the err out of the handler) up the stack, instead of processing it inside of the retryLoop

simply changing it to

return nil, nil

fixes the issue.

Whats odd (but perhaps separate) is that Recoverer middleware has no effect on actual panics in CQRS.

In addition, I observed some weird behavior with backoff, and taking a look at the code I don't see how it accounts for backoff time.

I am happy to submit a PR to change that line if its wrong, but I am afraid I don't quite understand how messages are handled and it could easily break something somewhere else.

func (g *GoChannel) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error) {
	g.closedLock.Lock()

	if g.closed {
		return nil, errors.New("Pub/Sub closed")
	}

	g.subscribersWg.Add(1)
	g.closedLock.Unlock()

pubsub_test.go

func TestPublishSubscribe_subscribe_closed(t *testing.T) {
	pubSub := gochannel.NewGoChannel(
		gochannel.Config{},
		watermill.NewStdLogger(true, true),
	)

	pubSub.Close()

	_, err := pubSub.Subscribe(context.Background(), "example.topic")
	if err != nil {
		fmt.Printf("err: %v\n", err)
	}
	_, err = pubSub.Subscribe(context.Background(), "example.topic")
	if err != nil {
		fmt.Printf("err: %v\n", err)
	}
}

An validation error is returned when I try to construct a publisher with a pre-constructed conn.

I've read the source code and found out that this method will validate the Connection field of amqp.Config, which makes me confused. According to the name of NewPublisherWithConnection, I think this method is no need to check the Connection field again since the connection is ready to use. Please have a look on it.

watermill-amqp version: 2.0.6

Hi

It looks like current AMQP pub/sub implementation doesn't re-create queues after reconnection. It cause errors in case of "exclusive" or "auto-delete" queues in RabbitMQ:

cannot consume from channel: Exception (404) Reason: "NOT_FOUND - no queue 'macsvc_reva2.orbitsoft_5065325975852691651' in vhost '/'

Here example of configuration which cause problems:

conf := amqp.Config{
    Connection: amqp.ConnectionConfig{
        AmqpURI:   "amqp://user:[email protected]:5672/"
    },
    Marshaler: amqp.DefaultMarshaler{},
    Exchange: amqp.ExchangeConfig{
        GenerateName: func(topic string) string {
            return "test_exchange"
        },
        Type:    "fanout",
        Durable: false,
    },
    Queue: amqp.QueueConfig{
        GenerateName: func(topic string) string {
            hostname, err := os.Hostname()
            if err != nil {
                hostname = "unknown"
            }

            return "mysvc_" + hostname + "_" + strconv.Itoa(rand.Int())
        },
        Exclusive:  true,
    },
    QueueBind: amqp.QueueBindConfig{
        GenerateRoutingKey: func(topic string) string {
            return ""
        },
    },
    Publish: amqp.PublishConfig{
        GenerateRoutingKey: func(topic string) string {
            return topic
        },
    },
    Consume: amqp.ConsumeConfig{
        Qos: amqp.QosConfig{
            PrefetchCount: 5,
        },
    },
    TopologyBuilder: &amqp.DefaultTopologyBuilder{},
}

Hey. Maybe it's not an issue, but is there a support for HTTP pub/sub with Google Cloud? What I mean is something like this:

https://cloud.google.com/run/docs/tutorials/pubsub

Would be nice to add a method to the Google Pub/Sub to return an HTTP server

Looking at the source code & examples I see that middlewares are executed top to bottom. In the comments here we see that Recoverer should happen before Retry. This should help folks that look at these examples.