tmq queues example

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

docs/examples/temporary_queues/tmp_queues.go

@@ -0,0 +1,150 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
+	"time"
+)
+
+func main() {
+
+	env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
+
+	// Open a connection to the AMQP 1.0 server ( RabbitMQ >= 4.0)
+	amqpConnection, err := env.NewConnection(context.Background())
+
+	if err != nil {
+		rmq.Error("Error opening connection", err)
+		return
+	}
+
+	// Create the management interface for the connection
+	// so we can declare exchanges, queues, and bindings
+	management := amqpConnection.Management()
+	_, err = management.DeclareExchange(context.Background(), &rmq.FanOutExchangeSpecification{
+		Name: "broadcast",
+	})
+
+	if err != nil {
+		rmq.Error("Error declaring exchange", err)
+		return
+	}
+
+	// Declare a temp queues
+
+	tmpQueueInfo1, err := management.DeclareQueue(context.Background(), &rmq.AutoGeneratedQueueSpecification{
+		IsAutoDelete: true,
+		IsExclusive:  true,
+	})
+
+	if err != nil {
+		rmq.Error("Error declaring queue", err)
+		return
+	}
+
+	go func() {
+		consumer, err := amqpConnection.NewConsumer(context.Background(), tmpQueueInfo1.Name(), nil)
+		if err != nil {
+			rmq.Error("Error creating consumer", err)
+			return
+		}
+		for {
+			dcx, err1 := consumer.Receive(context.Background())
+			if err1 != nil {
+				rmq.Error("Error receiving message", err)
+				return
+			}
+			rmq.Info("[Consumer tmp1] Received message", "msg", fmt.Sprintf("%s", dcx.Message().Data))
+			err1 = dcx.Accept(context.Background())
+			if err1 != nil {
+				rmq.Error("Error accepting message", err)
+				return
+			}
+		}
+	}()
+
+	tmpQueueInfo2, err := management.DeclareQueue(context.Background(), &rmq.AutoGeneratedQueueSpecification{
+		IsAutoDelete: true,
+		IsExclusive:  true,
+	})
+
+	if err != nil {
+		rmq.Error("Error declaring queue", err)
+		return
+	}
+
+	go func() {
+		consumer, err := amqpConnection.NewConsumer(context.Background(), tmpQueueInfo2.Name(), nil)
+		if err != nil {
+			rmq.Error("Error creating consumer", err)
+			return
+		}
+		for {
+			dcx, err1 := consumer.Receive(context.Background())
+			if err1 != nil {
+				rmq.Error("Error receiving message", err)
+				return
+			}
+			rmq.Info("[Consumer tmp2] Received message", "msg", fmt.Sprintf("%s", dcx.Message().Data))
+			err1 = dcx.Accept(context.Background())
+			if err1 != nil {
+				rmq.Error("Error accepting message", err)
+				return
+			}
+		}
+	}()
+
+	_, err = management.Bind(context.TODO(), &rmq.ExchangeToQueueBindingSpecification{
+		SourceExchange:   "broadcast",
+		DestinationQueue: tmpQueueInfo1.Name(),
+	})
+
+	if err != nil {
+		rmq.Error("Error binding", err)
+		return
+	}
+
+	_, err = management.Bind(context.TODO(), &rmq.ExchangeToQueueBindingSpecification{
+		SourceExchange:   "broadcast",
+		DestinationQueue: tmpQueueInfo2.Name(),
+	})
+
+	if err != nil {
+		rmq.Error("Error binding", err)
+		return
+	}
+
+	publisher, err := amqpConnection.NewPublisher(context.Background(), &rmq.ExchangeAddress{
+		Exchange: "broadcast",
+	}, nil)
+
+	if err != nil {
+		rmq.Error("Error creating publisher", err)
+		return
+	}
+
+	for i := 0; i < 10_000; i++ {
+		publishResult, err := publisher.Publish(context.Background(),
+			rmq.NewMessage([]byte("Hello AMQP 1.0 - id:"+fmt.Sprintf("%d", i))))
+		if err != nil {
+			rmq.Error("Error publishing message", err)
+			return
+		}
+
+		switch publishResult.Outcome.(type) {
+		// publish result
+		case *rmq.StateAccepted:
+			rmq.Info("Message accepted", "message", publishResult.Message.GetData())
+		default:
+			rmq.Error("Message not accepted", "outcome", publishResult.Outcome)
+		}
+		time.Sleep(1 * time.Second)
+	}
+
+	// press any key to close the connection
+
+	var input string
+	_, _ = fmt.Scanln(&input)
+
+}