Merge pull request #161 from TraGicCode/fix-rabbitmq-issue-on-shutdown

TraGicCode · web-flow · commit 7f39b5200e55 · 2025-12-18T10:57:05.000-06:00
Fix issue on graceful shutdown with rabbitmq due to not copying memory for the message body
diff --git a/src/BuslyCLI.Console/Commands/Command/SendCommand.cs b/src/BuslyCLI.Console/Commands/Command/SendCommand.cs
@@ -40,8 +40,7 @@ await rawEndpoint.Dispatch(
             new TransportTransaction()
         );
 
-        // TODO: Figure out why this causes connection closed errors with rabbitmq
-        // await rawEndpoint.ShutDownAndCleanUp();
+        await rawEndpoint.ShutDownAndCleanUp();
 
         return 0;
     }
diff --git a/src/BuslyCLI.Console/Commands/Event/PublishCommand.cs b/src/BuslyCLI.Console/Commands/Event/PublishCommand.cs
@@ -45,8 +45,7 @@ await rawEndpoint.Dispatch(
             new TransportTransaction()
         );
 
-        // TODO: Figure out why this causes connection closed errors with rabbitmq
-        // await rawEndpoint.ShutDownAndCleanUp();
+        await rawEndpoint.ShutDownAndCleanUp();
 
         return 0;
     }
diff --git a/src/BuslyCLI.Console/Factories/RawEndpoint.cs b/src/BuslyCLI.Console/Factories/RawEndpoint.cs
@@ -66,7 +66,7 @@ await _subscriptionManager.SubscribeAll([new MessageMetadata(CreateTypeFromStrin
     public Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken)
     {
         _receivedMessages.Add(
-            new IncomingMessage(messageContext.NativeMessageId, messageContext.Headers, messageContext.Body),
+            new IncomingMessage(messageContext.NativeMessageId, messageContext.Headers, messageContext.Body.ToArray()),
             cancellationToken);
         return Task.CompletedTask;
     }

Original file line number	Diff line number	Diff line change
`@@ -40,8 +40,7 @@ await rawEndpoint.Dispatch(`
`40`	`40`	`new TransportTransaction()`
`41`	`41`	`);`
`42`	`42`
`43`		`- // TODO: Figure out why this causes connection closed errors with rabbitmq`
`44`		`- // await rawEndpoint.ShutDownAndCleanUp();`
	`43`	`+ await rawEndpoint.ShutDownAndCleanUp();`
`45`	`44`
`46`	`45`	`return 0;`
`47`	`46`	`}`
Original file line number	Diff line number	Diff line change
`@@ -45,8 +45,7 @@ await rawEndpoint.Dispatch(`
`45`	`45`	`new TransportTransaction()`
`46`	`46`	`);`
`47`	`47`
`48`		`- // TODO: Figure out why this causes connection closed errors with rabbitmq`
`49`		`- // await rawEndpoint.ShutDownAndCleanUp();`
	`48`	`+ await rawEndpoint.ShutDownAndCleanUp();`
`50`	`49`
`51`	`50`	`return 0;`
`52`	`51`	`}`
Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ await _subscriptionManager.SubscribeAll([new MessageMetadata(CreateTypeFromStrin`
`66`	`66`	`public Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken)`
`67`	`67`	`{`
`68`	`68`	`_receivedMessages.Add(`
`69`		`- new IncomingMessage(messageContext.NativeMessageId, messageContext.Headers, messageContext.Body),`
	`69`	`+ new IncomingMessage(messageContext.NativeMessageId, messageContext.Headers, messageContext.Body.ToArray()),`
`70`	`70`	`cancellationToken);`
`71`	`71`	`return Task.CompletedTask;`
`72`	`72`	`}`