Merge branch 'SOL-62456' into EBP-18

internal/impl/message/inbound_message_impl.go

@@ -164,7 +164,7 @@ func (inboundMessage *InboundMessageImpl) GetCacheRequestID() (message.CacheRequ
 	cacheID, errInfo := ccsmp.SolClientMessageGetCacheRequestID(inboundMessage.messagePointer)
 	if errInfo != nil {
 		if errInfo.ReturnCode == ccsmp.SolClientReturnCodeFail {
-			logging.Default.Info(fmt.Sprintf("Encountered error retrieving Cache ID: %s, subcode: %d", errInfo.GetMessageAsString(), errInfo.SubCode))
+			logging.Default.Info(fmt.Sprintf("Encountered error retrieving Cache ID: %s, subcode: %d", errInfo.GetMessageAsString(), errInfo.SubCode()))
 		}
 		return 0, false
 	}

internal/impl/message/inbound_message_test.go

@@ -28,7 +28,7 @@ func TestGetCacheRequestID(t *testing.T) {
 		t.Error("did not expect error, got " + ccsmpErr.GetMessageAsString())
 	}
 	msg := NewInboundMessage(msgP, false)
-	if msg.messagePointer == nil {
+	if msg.messagePointer == ccsmp.SolClientOpaquePointerInvalidValue {
 		t.Error("expected message pointer to not be nil")
 	}
 	if msg.IsDisposed() {

test/endpoint_provisioner_test.go

@@ -24,6 +24,7 @@ import (
 	"solace.dev/go/messaging"
 	"solace.dev/go/messaging/pkg/solace"
 	"solace.dev/go/messaging/pkg/solace/config"
+	"solace.dev/go/messaging/pkg/solace/subcode"
 	"solace.dev/go/messaging/test/helpers"
 	"solace.dev/go/messaging/test/testcontext"
 
@@ -118,15 +119,23 @@ var _ = Describe("EndpointProvisioner", func() {
 	})
 
 	Context("with a connected messaging service that will be disconnected", func() {
+		var eventChannel chan solace.ServiceEvent
 		var provisioner solace.EndpointProvisioner
+		var listenerID uint64
 
 		BeforeEach(func() {
 			helpers.ConnectMessagingService(messagingService)
+			eventChannel = make(chan solace.ServiceEvent, 1)
+			listenerID = messagingService.AddServiceInterruptionListener(func(event solace.ServiceEvent) {
+				eventChannel <- event
+			})
+
 			provisioner = messagingService.EndpointProvisioner()
 			Expect(provisioner).ToNot(BeNil())
 		})
 
 		AfterEach(func() {
+			messagingService.RemoveServiceInterruptionListener(listenerID)
 			if messagingService.IsConnected() {
 				messagingService.Disconnect()
 			}
@@ -138,6 +147,11 @@ var _ = Describe("EndpointProvisioner", func() {
 			},
 			"SEMPv2 disconnect": func(messagingService solace.MessagingService) {
 				helpers.ForceDisconnectViaSEMPv2(messagingService)
+				var serviceEvent solace.ServiceEvent
+				Eventually(eventChannel, 20*time.Second).Should(Receive(&serviceEvent), "Expected to receive service down event")
+				Expect(time.Since(serviceEvent.GetTimestamp())).To(BeNumerically("<=", 1*time.Second), "Expected timestamp of service event to be recent")
+				helpers.ValidateNativeError(serviceEvent.GetCause(), subcode.CommunicationError)
+				Consistently(eventChannel).ShouldNot(Receive()) // should not have another event since messaging service is down
 			},
 		}
 
@@ -158,6 +172,7 @@ var _ = Describe("EndpointProvisioner", func() {
 
 			It("fails to deprovision with "+testCase, func() {
 				disconnectFunction(messagingService)
+				Eventually(messagingService.IsConnected(), 30*time.Second).Should(BeFalse())
 
 				err := provisioner.Deprovision(provisionQueueName, true)
 				Expect(err).To(HaveOccurred())

test/request_reply_message_receiver_test.go

@@ -227,7 +227,6 @@ var _ = Describe("RequestReplyReceiver", func() {
 					if replier != nil {
 						payload, _ := racingMessage.GetPayloadAsString()
 						err = replier.Reply(helpers.NewMessage(messagingService, "Reply for: "+payload))
-						Expect(err).To(BeNil())
 					}
 				}
 

version.go

@@ -23,4 +23,4 @@ func init() {
 	core.SetVersion(version)
 }
 
-const version = "1.8.0"
+const version = "1.8.1"