Do not work with RabbitMQ

[laibulle]

Hello,

I try to get this package work with RabbitMQ and the AMQP 1 plugin. I can get it work with Go and Azure or C# and RabbitMQ but not with Go and RabbitMQ. Here is my code

package main

import (
	"context"
	"time"

	log "github.com/sirupsen/logrus"
	"pack.ag/amqp"
)

const amqpURL = "amqp://xxxxx:xxxxx@localhost:5672/"

func main() {

	// Create client
	client, err := amqp.Dial(amqpURL)

	ctx := context.Background()

	if err != nil {
		log.Fatal(err)
	}

	// Open a session
	session, err := client.NewSession()
	if err != nil {
		log.Fatal("Creating AMQP session:", err)
	}

	// Create a sender
	sender, err := session.NewSender(
		amqp.LinkTargetAddress("/queue-name"),
	)

	if err != nil {
		log.Fatal("Creating sender link:", err)
	}

	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)

	// Send message
	msg := "Hello!"
	err = sender.Send(ctx, amqp.NewMessage([]byte(msg)))
	if err != nil {
		log.Fatal("Sending message:", err)
	}

	log.Info("sent %s", msg)

	cancel()
	sender.Close()
}

The code is blocked at session.NewSender and moreover I can't see the connection in RabbitMQ admin.

Did I miss something ?

Best regard

[laibulle]

FYI, I did not created the topic. So I found this log entry in

{'v1_0.error',{symbol,<<"amqp:invalid-field">>},{utf8,<<"Attach rejected: {unknown_destination,\"/queue-name\"}">>},undefined}

but the driver did not returned any error.

[vcabbage]

Thanks for the report!

The existing logic wasn't propagating session errors during the attachment. I've opened #61 to resolve this.

I confirmed this propagates the error. I was able to create a "queue-name" in RabbitMQ and attach using "/topic/queue-name" as the target address. It did hang when closing the sender due to the way RabbitMQ responds to the detach. I'll need to look into this a bit more to see if it's something that needs to be corrected in this library but don't have time at the moment. I hope to get to it soon though.

[vcabbage]

@laibulle The necessary fixes for this library to work with RabbitMQ have been completed. Closing a link will hang until https://github.com/rabbitmq/rabbitmq-amqp1.0/issues/60 is resolved, but it can be worked around by applying a timeout to the close via the context argument of Sender.Close or simply closing the client connection.

Please let me know if you run into any other problems. I'll leave this issue open at least until the RabbitMQ issue has been resolved.

[Nyarum]

@vcabbage this is still don't work. (Rabbit 3.7.7) Getting <nil> unexpected frame: &amqp.performClose{Error:(*amqp.Error)(0xc4202c41e0)} on create NewSession()

Test looks like:

// Connect to RabbitMQ
	amqpConnection, err = amqp.Dial(config.Mq.Rabbit.URL, amqp.ConnMaxFrameSize(5000))
	if err != nil {
		err = errors.Wrap(err, "Rabbit initialization failed")
		return
	}

	fmt.Println(amqpConnection.NewSession())

Can you help?

[vcabbage]

@Nyarum I tried to reproduce the issue with a RabbitMQ docker container with the AMQP 1.0 plugin enabled. I didn't get an error on NewSession().

Dockerfile:

FROM rabbitmq:3.7-management
RUN rabbitmq-plugins enable --offline rabbitmq_amqp1_0

Commands:

docker build -t rabbitmq-amqp10 .
docker run --rm -p 8080:15672 -p 5672:5672 rabbitmq-amqp10

If you modify: https://github.com/vcabbage/amqp/blob/271d6282f59cd10184fef147de589cce8ad69c93/conn.go#L376

in your local copy of the lib to be:

c.err = errorErrorf("unexpected frame: %v", fr.body)

It should print out the error information rather than the pointer, which may help track down the issue.