rabbitmq-amqp-go-client/docs/examples/web_sockets/web_sockets.go at 3524aeb704473373b78c2539f1cc1b08a84ab32d · rabbitmq/rabbitmq-amqp-go-client · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
// RabbitMQ AMQP 1.0 Go Client: https://github.com/rabbitmq/rabbitmq-amqp-go-client
// RabbitMQ AMQP 1.0 documentation: https://www.rabbitmq.com/docs/amqp
// The example is demonstrating how to connect to RabbitMQ using AMQP 1.0 over WebSocket protocol with SASLTypePlain,
// declare a queue, publish a message to it, and then consume that message.
// AMQP 1.0 over WebSocket documentation: https://www.rabbitmq.com/blog/2025/04/16/amqp-websocket
// example path: https://github.com/rabbitmq/rabbitmq-amqp-go-client/tree/main/docs/examples/web_sockets/web_sockets.go

package main

import (
	"context"

	"github.com/Azure/go-amqp"
	rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
	const amqpConnectionString = "ws://127.0.0.1:15678/ws"
	rmq.Info("[Example]", "Starting web socket connection to", amqpConnectionString)
	// for anonymous connection use:
	// env := rmq.NewEnvironment(amqpConnectionString, &rmq.AmqpConnOptions{
	// 	SASLType: amqp.SASLTypeAnonymous(),
	// })
	env := rmq.NewEnvironment(amqpConnectionString, &rmq.AmqpConnOptions{
		SASLType: amqp.SASLTypePlain("rabbit", "rabbit"),
	})
	conn, err := env.NewConnection(context.Background())
	if err != nil {
		panic(err)
	}
	_, err = conn.Management().DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{
		Name: "test-ws-queue",
	})
	if err != nil {
		panic(err)
	}
	// declare new producer
	producer, err := conn.NewPublisher(context.TODO(), &rmq.QueueAddress{
		Queue: "test-ws-queue",
	}, nil)
	if err != nil {
		panic(err)
	}
	msg := rmq.NewMessage([]byte("Hello over WebSockets"))

	publishResult, err := producer.Publish(context.Background(), msg)
	if err != nil {
		panic(err)
	}
	switch publishResult.Outcome.(type) {
	case *rmq.StateAccepted:
		rmq.Info("[Publisher]", "Message accepted", publishResult.Message.Data[0])
	default:
		rmq.Warn("[Publisher]", "Message not accepted", publishResult.Message.Data[0])
	}

	// declare new consumer
	consumer, err := conn.NewConsumer(context.TODO(), "test-ws-queue", nil)
	if err != nil {
		panic(err)
	}
	deliveryContext, err := consumer.Receive(context.Background())
	if err != nil {
		panic(err)
	}
	rmq.Info("[Consumer]", "Message received", string(deliveryContext.Message().GetData()))
	err = deliveryContext.Accept(context.Background())
	if err != nil {
		panic(err)
	}
	// clean up
	err = consumer.Close(context.TODO())
	if err != nil {
		panic(err)
	}
	err = producer.Close(context.TODO())
	if err != nil {
		panic(err)
	}

	err = conn.Management().DeleteQueue(context.Background(), "test-ws-queue")
	if err != nil {
		panic(err)
	}

	err = conn.Close(context.TODO())
	if err != nil {
		panic(err)
	}

}