When a message was sent and it's response was received while the connection was closed or an error happened, the reader go-routine could get stuck and be leaked.

The reader go routine tries to send a received message to the unbuffered c.rpc channel via the dispatch0() and dispatchN() methods. The call() method reads from the rpc channel. If an error happened while the dispatch method sends a message to the rpc channel, the call() method could terminate because it read an error from c.errors or because c.errors was closed.

To prevent the scenario:

• the reader go-routine now closes c.rpc when it terminates,
• The call() method, reads from c.rpc until a message was received or it is closed. When c.rpc is closed, it reads an error from c.errors or wait until c.errors is closed. When it reads an error, it returns it. If it is closed it returns ErrClosed.

This ensures that the messages is read from c.rpc before call() returns. It also ensures that when a message was received that it is processed. Previously it could happen that the message was silently ignored because c.errors returned an error or was closed.

This fixes #69. A testcase can be found in PR https://github.com/rabbitmq/amqp091-go/pull/71

Adds a new method, PublishAndAwaitConfirm, to channels which allows users to publish a message and wait on its confirmation.

There are two problems with the existing mechanisms, NotifyPublish and NotifyConfirm. The first is that they both only expose confirmations to the user in-order, making it impossible to wait for a single confirmation without also waiting for every previous confirmation. The second is that the way they relay the confirmation back to the user is difficult to correlate to the published message. They expose the delivery tag through the Confirmation struct, but the delivery tag of the publish is not exposed, forcing the user to implement their own delivery tag accounting in order to correlate a publish and a confirm.

This implementation essentially returns a future for the confirmation from the PublishAndAwaitConfirm call, making it easy for users to correlate the publish to the confirm and hiding all of the delivery tag accounting to inside the library.

cc @ChunyiLyu @ikvmw

Sample usage:

ch.Confirm(false)
confirm, _ := ch.ConfirmPublish(...)
if !confirm.Wait() {
  log.Fatal("publish was nacked!")
}

One remaining question I have is if we want to deprecate NotifyPublish/NotifyConfirm. I believe that it'd be pretty straightforward to build in-order confirmations on top of this, so it's a superset of those APIs.

I'm using NotifyPublish to get confirms. At the end of the run, I'm trying to close the connection and the associate channel, but it hangs indefinitely.

Here's a test to reproduce the issue (reproduces 80-90% of the time):

func TestCloseHandBug(t *testing.T) {
	ctx := context.Background()
	testQueue := "test"

	c, err := amqp091.Dial(fmt.Sprintf("amqp://%s:%s@%s%s", Config.Username, Config.Password, Config.Hostname, Config.VHost))
	if err != nil {
		t.Fatalf("Error connecting to server: %s", err)
	}
	t.Log("connected")

	closeChan := make(chan struct{})

	// close connection
	defer func() {
		close(closeChan)
		if c != nil {
			t.Log("disconnecting")
			//TODO: program hangs here - removing c.close and adding
			//TODO: <-time.After(time.Second * 120) doesn't help. closing channel hangs
			if err := c.Close(); err != nil {
				t.Logf("disconnect error: %s", err)
				return
			}
		}

		c = nil
		t.Log("disconnected")
	}()

	ch, err := c.Channel()
	if err != nil && err == amqp091.ErrClosed {
		t.Fatalf("couldn't open channel: %s", err)
	}

	err = ch.Qos(
		16,    // prefetch count
		0,     // prefetch size
		false, // global
	)
	if err != nil {
		t.Fatalf("couldn't configure qos: %s", err)
	}

	// handle confirms TODO: without this, disconnection works
	confirms := ch.NotifyPublish(make(chan amqp091.Confirmation))
	go func() {
		for {
			select {
			case <-confirms:
				// do something here in real scenario
				continue
			case <-closeChan:
				return
			case <-ctx.Done():
				return
			}
		}
	}()

	err = ch.Confirm(false)
	if err != nil {
		t.Fatalf("couldn't configure confirm: %s", err)
	}

	ch.ExchangeDeclare("amq.topic", "topic", true, false, false, false, nil)

	// handle close channel TODO: removing this func doesn't solve the issue
	go func() {
		for {
			select {
			case <-closeChan:
				t.Logf("closing channel. id=%d",1)
				//TODO: program hangs here as well
				if err := ch.Close(); err != nil {
					t.Logf("closing channel err. id=%d. err=%d",1, err)
					return
				}
				t.Logf("channel closed. id=%d",1)
				return
			case <-ctx.Done():
				return
			}
		}
	}()

	_, err = ch.QueueDelete(testQueue, false, false, false)
	if err != nil {
		t.Fatalf("error deleting test queue: %s", err)
	}

	key := "subscribe.send"

	_, err = ch.QueueDeclare(testQueue, true, false, false, false, nil)
	if err != nil {
		t.Fatalf("error declaring queue: %s", err)
	}

	err = ch.QueueBind(testQueue, key, "amq.topic", false, nil)
	if err != nil {
		t.Fatalf("error binding queue: %s", err)
	}

	delivery, err := ch.Consume(testQueue, "", false, false, false, false, nil)
	if err != nil {
		t.Fatalf("error consuming: %s", err)
	}

	expected := "hello world!"

	t.Logf("sending. id=%d, data=%s",1, []byte(expected))

	m := amqp091.Publishing{
		ContentType:  "text/plain",
		Body:         []byte(expected),
		DeliveryMode: amqp091.Persistent,
	}

	err = ch.Publish("amq.topic", key, true, false, m)
	if err != nil {
		t.Fatalf("Error publishing: %s", err)
	}

	var actual string
	select {
	case message := <-delivery:
		message.Ack(false)
		actual = string(message.Body)
	case <-time.After(time.Second / 2):
		t.Fatalf("Timeout recieving message")
	}

	if actual != expected {
		t.Fatalf("Expected message %s, but got %s", expected, actual)
	}
}

Here is the result: === RUN TestCloseHandBug provider_test.go:125: connected provider_test.go:224: sending. id=1, data=hello world! provider_test.go:133: disconnecting provider_test.go:187: closing channel. id=1

• One goroutine waits on l <- confirmation in func (c *confirms) confirm(confirmation Confirmation)
• another goroutine waits on c.destructor.Do(func() { in func (c *Connection) shutdown(err *Error), originating in connection.close
• and another goroutine waits in the same place: c.destructor.Do(func() { in func (c *Connection) shutdown(err *Error), originating in channel.close
• another goroutine waits on c.m.Lock() in func (c *confirms) Close() error (originating in go c.shutdown(&Error{)

Notes:

1. If I remove the "handle close channel" function, the issue still occurs.
2. If I remove the c.close() part in the "close connection" section, waiting instead for a long time before closing the test, I will still never get to the "channel closed" part - it looks like closing the channel hangs indefinitely in this case as well.
3. Sometimes it works fine (10-20% of the time): === RUN TestCloseHandBug provider_test.go:125: connected provider_test.go:225: sending. id=1, data=hello world! provider_test.go:133: disconnecting provider_test.go:187: closing channel. id=1 provider_test.go:142: disconnected --- PASS: TestCloseHandBug (1.67s) PASS

This is a simplified version of my code.

package main

import (
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
)

func main() {
	amqpUri := "amqpuri"
	conn, err := amqp.Dial(amqpUri)
	if err != nil {
		return
	}
	notifyConnClose := make(chan *amqp.Error)
	conn.NotifyClose(notifyConnClose)
	log.Println("RabbitMQ client connected")

	ch, err := conn.Channel()
	if err != nil {
		return
	}
	notifyChanClose := make(chan *amqp.Error)
	ch.NotifyClose(notifyChanClose)

	queueName := "testqueue"
	_, err = ch.QueueDeclare(queueName, false, false, false, false, nil)
	if err != nil {
 		return
	}

	deliveryChan, err := ch.Consume(queueName, "", false, false, false, false, nil)
	if err != nil {
		return
    	}

	go func() {
	select {
	case <-notifyConnClose:
		log.Println("connection closed")
	case <-notifyChanClose:
		log.Println("channel closed")
	}
	}()

	for d := range deliveryChan {
	log.Println(string(d.Body))
	d.Ack(false)
	//ch.Close()  //comment out to test graceful close.
	}

	log.Println("terminating...")
}

In this code I am getting a connection and a channel and registering a notification (go)channel for both the connection and channel to be notified when they are closed.

Then I declare a queue and start consuming messages from it by ranging on the deliveryChan <-chan amqp.Delivery returned by the consume function.

The problem happens when an unexpected disconnection occurs (for example I turn off my internet) . In that case even though the notifyConnClose channel gets a message the deliveryChan is not closed, and the range loop blocks forever.

In the event of a graceful disconnection by a connection.Close() then both the notifyConnClose gets a message, and the deliveryChan is Closed.

In the event of the unexpected disconnection, given that I can't close the <-chan amqp.Delivery from my code how am I supposed to proceed and get the loop to end?

[Issue]

I have 2 single active consumers (A, B).

A first consumer was active, B second was ready.

then, i killed A first active consumer while other process publish 10000 messages.

I was expecting a transition from A to B for ACTIVE consumer.

but it isn't.

[Reproduce]

1. run rabbitmq server

docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmqtest rabbitmq:3.10.6-management

2. run consumer

func main() {
	conn, err := amqp091.DialConfig("amqp://guest:guest@localhost:5672/", amqp091.Config{
		Heartbeat: time.Second * 30,
	})
	if err != nil {
		panic(err)
	}

	ch, err := conn.Channel()
	if err != nil {
		panic(err)
	}

	if err := ch.Qos(200, 0, false); err != nil {
		panic(err)
	}

	queueArgs := make(amqp091.Table)
	queueArgs["x-single-active-consumer"] = true
	_, err = ch.QueueDeclare("queue",
		true,      // durable
		false,     // auto delete
		false,     //exclusive
		false,     //noWait
		queueArgs, // queue args
	)
	if err != nil {
		panic(err)
	}

	msgs, err := ch.Consume("queue", "consumer", false, false, false, false, nil)
	if err != nil {
		panic(err)
	}
	d := make(chan bool)
	go func() {
		for msg := range msgs {
			fmt.Println(string(msg.Body))
			_ = msg.Ack(true)
		}
		d <- true
	}()

	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGKILL)
	<-sigs

	if err := ch.Cancel("consumer", false); err != nil {
		panic(err)
	}
	fmt.Println("cancel consume")
	if err := ch.Close(); err != nil {
		panic(err)
	}
	if err := conn.Close(); err != nil {
		panic(err)
	}
	<-d
	fmt.Println("terminate")
}

go run consumer.go (A session)
go run consumer.go (B session)

3. run producer

func main() {
	conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		panic(err)
	}

	ch, err := conn.Channel()
	if err != nil {
		panic(err)
	}

	for i := 0; i < 10000; i++ {
		err := ch.Publish("", "queue", false, false, amqp091.Publishing{
			DeliveryMode: 0,
			ContentType:  "text/plain",
			Body:         []byte(fmt.Sprintf("%d", i)),
		})
		if err != nil {
			panic(err)
		}
	}

	if err := ch.Close(); err != nil {
		panic(err)
	}
	if err := conn.Close(); err != nil {
		panic(err)
	}
	fmt.Println("terminate")
}

4. check active consumer

root@2d6192fe4fed:/# rabbitmqctl list_consumers
Listing consumers in vhost / ...
queue_name	channel_pid	consumer_tag	ack_required	prefetch_count	active	arguments
queue	<[email protected]>	consumer	true	200	false	[]
queue	<[email protected]>	consumer	true	200	true	[]

5. run publisher

go run publisher.go

6. kill Active consumer when message is received. It is important that Active consumer does not receive all messages. It should be killed in the middle of receiving messages.

# active consumer log
1
2
3
4
...
^C (it will be print text "terminate")

7. check active consumer

root@2d6192fe4fed:/# rabbitmqctl list_consumers
Listing consumers in vhost / ..

As a result, I was expecting a transition from A to B , but it didn't. All consumers are gone.

However, I found error logs of rabbitmq server. It's too long, so I'm attaching it as a file. rabbit-error-logs.txt

It doesn't seem to be a problem with amqp091-go.

Hi, I already covered created an issue for this on https://github.com/streadway/amqp/issues/518 but then it was for streadway/amqp the issue still reproduces with this library so I decided to add the issue here too

I have created a sample project with few files that you have to download and run the main to reproduce the issue, the files are on: https://gist.github.com/melardev/1b9c7e1b1a4ac37cb31e57dc6cde99c7

The code is bad, it is not thread-safe, but It was easier to reproduce my issue with that code rather than my real project that was doing effort to keep things thread-safe, the idea is to show the framework hanging yet the connection is already closed, it may get stuck in any function call, mainly QueueDeclare and channel.Close() but most of the times it gets stuck on channel.Close() the issue is for closing a channel it has to acquire a Lock https://github.com/rabbitmq/amqp091-go/blob/main/connection.go#L640, which is already held here https://github.com/rabbitmq/amqp091-go/blob/main/connection.go#L408, basically the connection is waiting the channel to be closed but the channel::shutdown() never exits its main loop, even if the connection is closed ...

Implement #92

~This currently require context to be always cancelled (like net/http query) to not have stuck goroutine.~

~I am considering switching from waitgroup to the context of query for the sync pattern of DeferredConfirmation. This could lead to no leaking goroutine.~

The latest implementation replace waitgroup of deferred with context so that we don't need goroutine to monitor context.

I use Rabbitmq3.11.3, and the official say "Single active consumer for streams is a feature available in RabbitMQ 3.11 and more." but when I use amqp091-go to declare a queue which type is steam,just like this: queue, err := channel.QueueDeclare( "q5", true, false, false, false, amqp091.Table{ "x-single-active-consumer": true, "x-queue-type": "stream", }, ) it always report error： 2022/11/23 16:56:19 Failed to declare a queue: Exception (406) Reason: "PRECONDITION_FAILED - invalid arg 'x-single-active-consumer' for queue 'q5' in vhost '/' of queue type rabbit_stream_queue" How can I do for it?

ci: remove travis ci config

Travis is not used anymore, github actions is used instead

-------------------------------------------------------------------------------
ci: run tests via make

Run the tests in github actions via "make tests" instead of duplicating
the command to run the tests.

This makes it easier to run the tests the same way locally and keep the
commands in sync.

The makefile tests target is changed to pass "-cpu=1,2", to run the
tests in CI the same way then before.

example_client_test.go shows a great way how to build a Producer with reconnection support.

The Consuming part is not covering the reconnection support: https://github.com/rabbitmq/amqp091-go/blob/51fade522e4d209b3cef22ff8c052d944524ac17/example_client_test.go#L257-L274

Even if reconnection support is mentioned as a non-goal for the library itself, I do think showcasing this in the example might be a good thing.

Right now, the implementation is only possible with a hack involved. Mainly due to these two issues:

• https://github.com/rabbitmq/amqp091-go/issues/18
• https://github.com/rabbitmq/amqp091-go/issues/32

These deadlocks prevent the consumer delivery channel to be closed during an unexpected connection loss (e.g., when the RabbitMQ server goes down).

Hi,

When writing persistent messages using this library, the messages are being written to "queues" folder but while using php amqp library, it is being written to "msg_store_persistent" folder. Where is this configuration being set? This is not an issue with the library but I need help to figure this out, thanks.

Calling Channel() on an uninitialized connection panics, rather than return an error.

Code:

package main

import (
	"fmt"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// panics instead of errors
	conn := &amqp.Connection{}
	_, err := conn.Channel()
	fmt.Println(err.Error())
}

Expected Behavior: prints out an error

Actual Behavior: panics

Invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x8 pc=0x4eec5d]

goroutine 1 [running]:
github.com/rabbitmq/amqp091-go.(*allocator).next(0x0)
	/tmp/gopath3681466631/pkg/mod/github.com/rabbitmq/[email protected]/allocator.go:72 +0x1d
github.com/rabbitmq/amqp091-go.(*Connection).allocateChannel(0xc000002140)
	/tmp/gopath3681466631/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:686 +0xd2
github.com/rabbitmq/amqp091-go.(*Connection).openChannel(0x405a19?)
	/tmp/gopath3681466631/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:709 +0x25
github.com/rabbitmq/amqp091-go.(*Connection).Channel(...)
	/tmp/gopath3681466631/pkg/mod/github.com/rabbitmq/[email protected]/connection.go:736
main.main()
	/tmp/sandbox2002244425/prog.go:14 +0x2b

See https://go.dev/play/p/sseC5Beiapd

At the moment of this writing, supported Go version are 1.18 and 1.19. Golang has a support policy of N-1. From their docs:

Each major Go release is supported until there are two newer major releases.

One benefit of playing this story is a reduced go.sum file, which in turn will "fix" CVEs in security scanners. Note these scanners would report CVEs in old EOL'ed Go versions, the client library code is fine.

In Go 1.18 we can also replace interface{} with any, and explore the option of using generic functions.

This change removes embedded context.Context (which is generally an anti-pattern) from DeferredConfirmation. Instead, we use a simple channel to wait for ack/nack status. This approach is more flexible since it can be combined with timers, tickers, other contexts and channels in general using select{} statements and there is no overhead from context cancellation setup.

Note that #96 introduced a behavior where Wait would unblock and return false once the context passed to Publish expires. This commit reverts this (arguably breaking) behavior in favor of WaitContext function.

https://www.digitalocean.com/community/tutorials/how-to-use-contexts-in-go

https://github.com/rabbitmq/amqp091-go/discussions/121#discussioncomment-3788391

See the following:

• #96
• #122
• #103

cc @mgdotson

Hello

package main

import (
	"context"
	"github.com/rabbitmq/amqp091-go"
	"sync"
	"time"
)

var wg sync.WaitGroup

func amqp(ctx context.Context) {
	defer func() {
		wg.Done()
	}()
	c, err := amqp091.Dial("amqp://guest:[email protected]:5672/")
	if err != nil {
		panic("connection error")
	}
	defer c.Close()

	<-ctx.Done()
}

const n = 16

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	wg.Add(n)
	for i := 0; i < n; i++ {
		go amqp(ctx)
	}

	cancel()
	wg.Wait()

	time.Sleep(time.Hour)
}

The above code causes 100% CPU usage after a short while The problem does not always occur, so you may have to run it several times to reproduce the problem Strace attached after the problem occurred:

strace: Process 3167635 attached
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
100,00    4,571892     4571892         1         1 futex
  0,00    0,000010           5         2           rt_sigprocmask
  0,00    0,000002           1         2           getpid
  0,00    0,000001           0         2           gettid
  0,00    0,000001           0         2           tgkill
  0,00    0,000000           0         1           rt_sigaction
------ ----------- ----------- --------- --------- ----------------
100.00    4,571906                    10         1 total