test

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

.ci/ubuntu/gha-setup.sh

@@ -85,7 +85,7 @@ function start_rabbitmq
         --network "$docker_network_name" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/enabled_plugins:/etc/rabbitmq/enabled_plugins" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro" \
-         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/definitions.json:/etc/rabbitmq/definitions.json:ro" \
+        --volume "$GITHUB_WORKSPACE/.ci/ubuntu/definitions.json:/etc/rabbitmq/definitions.json:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/advanced.config:/etc/rabbitmq/advanced.config:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/certs:/etc/rabbitmq/certs:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/log:/var/log/rabbitmq" \

docs/examples/async_confirmation/async_confirmation

@@ -10,7 +10,12 @@ import (
 
 func main() {
 
-	env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
+	// create 10 million UUIDs
+
+	env := rmq.NewEnvironment("amqp://test:test@192.168.1.124:5672/", &rmq.AmqpConnOptions{
+
+		MaxFrameSize: 2048,
+	})
 
 	// Open a connection to the AMQP 1.0 server ( RabbitMQ >= 4.0)
 	amqpConnection, err := env.NewConnection(context.Background())
@@ -20,7 +25,7 @@ func main() {
 	}
 
 	amqpConnection.Management().DeleteQueue(context.Background(), "test")
-	amqpConnection.Management().DeclareQueue(context.Background(), &rmq.QuorumQueueSpecification{
+	amqpConnection.Management().DeclareQueue(context.Background(), &rmq.StreamQueueSpecification{
 		Name: "test",
 	})
 
@@ -38,16 +43,23 @@ func main() {
 
 		switch state.(type) {
 		case *amqp.StateAccepted:
-			if atomic.AddInt32(&confirmed, 1)%20_000 == 0 {
+			if atomic.AddInt32(&confirmed, 1)%50_000 == 0 {
 				// confirmations per second
 				rmq.Info("Confirmations per second", "value", float64(confirmed)/time.Since(startDate).Seconds())
 			}
+
+			if atomic.LoadInt32(&confirmed) == 2_500_000 {
+				// time since the start
+				rmq.Info("Time to confirm all messages", "value", time.Since(startDate).Seconds())
+			}
 		default:
 			panic("Message not accepted")
 
 		}
 
 	}
+	// 1kb to bytes
+	//bytes := make([]byte, 1_000)
 
 	//for i := 0; i < 500_000; i++ {
 	//	_, err := publisher.Publish(context.Background(), rmq.NewMessage(make([]byte, 1)))
@@ -58,14 +70,19 @@ func main() {
 	//		// message per second
 	//		rmq.Info("Sync Messages per second", "value", float64(i)/time.Since(startDate).Seconds())
 	//	}
+	//
+	//	if i == 500_000-1 {
+	//		// time since the start
+	//		rmq.Info("Time to confirm all messages", "value", time.Since(startDate).Seconds())
+	//	}
 	//}
 
-	for i := 0; i < 500_000; i++ {
-		err := publisher.PublishAsyncConfirmation(context.Background(), rmq.NewMessage(make([]byte, 1)), f)
+	for i := 0; i < 2_500_000; i++ {
+		err := publisher.PublishAsyncConfirmation(context.Background(), rmq.NewMessage(make([]byte, 1_000)), f)
 		if err != nil {
 			rmq.Error("Error publishing message", err)
 		}
-		if i%20_000 == 0 {
+		if i%50_000 == 0 {
 			// message per second
 			rmq.Info("Messages per second", "value", float64(i)/time.Since(startDate).Seconds())
 		}

docs/examples/async_confirmation/async_confirmation.go

@@ -38,7 +38,7 @@ func newPublisher(ctx context.Context, connection *AmqpConnection, destinationAd
 
 	r := &Publisher{connection: connection, linkName: getLinkName(options),
 		destinationAdd: destinationAdd, id: id,
-		maxInFlight: 10_00,
+		maxInFlight: 1_000,
 		condition:   sync.NewCond(&sync.Mutex{})}
 	connection.entitiesTracker.storeOrReplaceProducer(r)
 	err := r.createSender(ctx)
@@ -121,22 +121,24 @@ func (m *Publisher) Publish(ctx context.Context, message *amqp.Message) (*Publis
 type CallbackConfirmation func(message *amqp.Message, state DeliveryState, err error)
 
 func (m *Publisher) PublishAsyncConfirmation(ctx context.Context, message *amqp.Message, callback CallbackConfirmation) error {
-	//go func() {
-	r, err := m.sender.Load().SendWithReceipt(ctx, message, nil)
-	if err != nil {
-		return err
-	}
-	if atomic.AddInt32(&m.pending, 1) == int32(m.maxInFlight) {
+	if atomic.AddInt32(&m.pending, 1) >= int32(m.maxInFlight) {
 		m.condition.L.Lock()
 		m.condition.Wait()
 		m.condition.L.Unlock()
 	}
-	go func() {
-		state, err := r.Wait(ctx)
+
+	//go func() {
+	sendReceipt, err := m.sender.Load().SendWithReceipt(ctx, message, nil)
+	if err != nil {
+		return err
+	}
+
+	go func(sr amqp.SendReceipt) {
+		state, err := sr.Wait(ctx)
 		atomic.AddInt32(&m.pending, -1)
-		//m.condition.Signal()
+		m.condition.Signal()
 		callback(message, state, err)
-	}()
+	}(sendReceipt)
 	//}()
 	return nil
 }