Client Side Tracing & OAuth implementation

quarkus-solace-client/deployment/pom.xml

@@ -27,6 +27,10 @@
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-devservices-deployment</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-oidc-client-deployment</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-junit5-internal</artifactId>

quarkus-solace-client/deployment/src/main/java/com/solace/quarkus/deployment/SolaceProcessor.java

@@ -10,15 +10,14 @@
 
 import com.solace.messaging.MessagingService;
 import com.solace.quarkus.MessagingServiceClientCustomizer;
+import com.solace.quarkus.runtime.OidcProvider;
 import com.solace.quarkus.runtime.SolaceConfig;
 import com.solace.quarkus.runtime.SolaceRecorder;
 import com.solace.quarkus.runtime.observability.SolaceMetricBinder;
 import com.solacesystems.jcsmp.JCSMPFactory;
 
 import io.quarkus.arc.SyntheticCreationalContext;
-import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
-import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
-import io.quarkus.arc.deployment.UnremovableBeanBuildItem;
+import io.quarkus.arc.deployment.*;
 import io.quarkus.deployment.annotations.*;
 import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.ExtensionSslNativeSupportBuildItem;
@@ -31,13 +30,14 @@
 import io.quarkus.smallrye.health.deployment.spi.HealthBuildItem;
 
 class SolaceProcessor {
-
     private static final String FEATURE = "solace-client";
 
     private static final ParameterizedType SOLACE_CUSTOMIZER_INJECTION_TYPE = ParameterizedType.create(
             DotName.createSimple(Instance.class),
             new Type[] { ClassType.create(DotName.createSimple(MessagingServiceClientCustomizer.class.getName())) }, null);
 
+    private static final Type OIDC_PROVIDER = ClassType.create(DotName.createSimple(OidcProvider.class));
+
     private static final AnnotationInstance[] EMPTY_ANNOTATIONS = new AnnotationInstance[0];
 
     @BuildStep
@@ -59,21 +59,24 @@ ExtensionSslNativeSupportBuildItem ssl() {
     @Record(ExecutionTime.RUNTIME_INIT)
     ServiceStartBuildItem init(
             SolaceConfig config, SolaceRecorder recorder,
-            ShutdownContextBuildItem shutdown, BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+            ShutdownContextBuildItem shutdown, BuildProducer<SyntheticBeanBuildItem> syntheticBeans,
+            BuildProducer<AdditionalBeanBuildItem> additionalBeanBuildItemBuildProducer) {
 
         Function<SyntheticCreationalContext<MessagingService>, MessagingService> function = recorder.init(config, shutdown);
 
+        additionalBeanBuildItemBuildProducer.produce(AdditionalBeanBuildItem.unremovableOf(OidcProvider.class));
+
         SyntheticBeanBuildItem.ExtendedBeanConfigurator solaceConfigurator = SyntheticBeanBuildItem
                 .configure(MessagingService.class)
                 .defaultBean()
                 .scope(ApplicationScoped.class)
                 .addInjectionPoint(SOLACE_CUSTOMIZER_INJECTION_TYPE, EMPTY_ANNOTATIONS)
+                .addInjectionPoint(OIDC_PROVIDER)
                 .createWith(function)
                 .unremovable()
                 .setRuntimeInit();
 
         syntheticBeans.produce(solaceConfigurator.done());
-
         return new ServiceStartBuildItem(FEATURE);
     }
 

quarkus-solace-client/runtime/pom.xml

@@ -18,7 +18,10 @@
       <groupId>com.solace</groupId>
       <artifactId>solace-messaging-client</artifactId>
     </dependency>
-
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-oidc-client</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-micrometer</artifactId>

quarkus-solace-client/runtime/src/main/java/com/solace/quarkus/runtime/OidcProvider.java

@@ -0,0 +1,59 @@
+package com.solace.quarkus.runtime;
+
+import java.time.Duration;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.event.Observes;
+import jakarta.inject.Inject;
+
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+
+import com.solace.messaging.MessagingService;
+import com.solace.messaging.config.SolaceProperties;
+
+import io.quarkus.oidc.client.OidcClient;
+import io.quarkus.oidc.client.Tokens;
+import io.quarkus.runtime.StartupEvent;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.infrastructure.Infrastructure;
+
+@ApplicationScoped
+public class OidcProvider {
+
+    @ConfigProperty(name = "quarkus.solace.oidc.refresh.interval", defaultValue = "60s")
+    Duration duration;
+
+    @Inject
+    OidcClient client;
+
+    private volatile Tokens lastToken;
+    private MessagingService service;
+
+    Tokens getToken() {
+        Tokens firstToken = client.getTokens().await().indefinitely();
+        lastToken = firstToken;
+        return firstToken;
+    }
+
+    void init(MessagingService service) {
+        this.service = service;
+    }
+
+    void startup(@Observes StartupEvent event) {
+        Multi.createFrom().ticks().every(duration)
+                .emitOn(Infrastructure.getDefaultWorkerPool())
+                //                .filter(aLong -> {
+                //                    if (lastToken.isAccessTokenWithinRefreshInterval()) {
+                //                        return true;
+                //                    } else
+                //                        return false;
+                //                })
+                .call(() -> client.getTokens().invoke(tokens -> {
+                    lastToken = tokens;
+                }))
+                .invoke(() -> service.updateProperty(SolaceProperties.AuthenticationProperties.SCHEME_OAUTH2_ACCESS_TOKEN,
+                        lastToken.getAccessToken()))
+                .subscribe().with(aLong -> {
+                });
+    }
+}

quarkus-solace-client/runtime/src/main/java/com/solace/quarkus/runtime/SolaceRecorder.java

@@ -38,11 +38,18 @@ public MessagingService apply(SyntheticCreationalContext<MessagingService> conte
                     }
                 }
 
-                MessagingServiceClientBuilder builder = MessagingService.builder(ConfigurationProfile.V1)
-                        .fromProperties(properties);
-
                 Instance<MessagingServiceClientCustomizer> reference = context.getInjectedReference(CUSTOMIZER);
+                OidcProvider oidcProvider = context.getInjectedReference(OidcProvider.class);
 
+                String authScheme = config.extra().get("authentication.scheme");
+
+                if (oidcProvider != null && authScheme != null && authScheme.equals("AUTHENTICATION_SCHEME_OAUTH2")) {
+                    properties.put(SolaceProperties.AuthenticationProperties.SCHEME_OAUTH2_ACCESS_TOKEN,
+                            oidcProvider.getToken().getAccessToken());
+                }
+
+                MessagingServiceClientBuilder builder = MessagingService.builder(ConfigurationProfile.V1)
+                        .fromProperties(properties);
                 MessagingService service;
                 if (reference.isUnsatisfied()) {
                     service = builder.build();
@@ -54,12 +61,14 @@ public MessagingService apply(SyntheticCreationalContext<MessagingService> conte
                     }
                 }
 
+                oidcProvider.init(service);
                 var tmp = service;
                 shutdown.addLastShutdownTask(() -> {
                     if (tmp.isConnected()) {
                         tmp.disconnect();
                     }
                 });
+
                 return service.connect();
             }
         };

quarkus-solace-messaging-connector/runtime/pom.xml

@@ -23,6 +23,11 @@
             <groupId>io.smallrye.reactive</groupId>
             <artifactId>smallrye-connector-attribute-processor</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>smallrye-reactive-messaging-otel</artifactId>
+            <version>4.16.0</version>
+        </dependency>
         <dependency>
             <groupId>com.solace</groupId>
             <artifactId>solace-messaging-client</artifactId>
@@ -132,6 +137,13 @@
             <artifactId>slf4j-log4j12</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.keycloak</groupId>
+            <artifactId>keycloak-admin-client</artifactId>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
     <build>

quarkus-solace-messaging-connector/runtime/src/main/java/com/solace/quarkus/messaging/SolaceConnector.java

@@ -37,6 +37,7 @@
 //@ConnectorAttribute(name = "client.type", type = "string", direction = INCOMING_AND_OUTGOING, description = "Direct or persisted", defaultValue = "persisted")
 @ConnectorAttribute(name = "client.lazy.start", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Whether the receiver or publisher is started at initialization or lazily at subscription time", defaultValue = "false")
 @ConnectorAttribute(name = "client.graceful-shutdown", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Whether to shutdown client gracefully", defaultValue = "true")
+@ConnectorAttribute(name = "client.tracing-enabled", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Whether to enable tracing for incoming and outgoing messages", defaultValue = "false")
 @ConnectorAttribute(name = "client.graceful-shutdown.wait-timeout", type = "long", direction = INCOMING_AND_OUTGOING, description = "Timeout in milliseconds to wait for messages to finish processing before shutdown", defaultValue = "10000")
 @ConnectorAttribute(name = "consumer.queue.name", type = "string", direction = INCOMING, description = "The queue name of receiver.")
 @ConnectorAttribute(name = "consumer.queue.type", type = "string", direction = INCOMING, description = "The queue type of receiver. Supported values `durable-exclusive`, `durable-non-exclusive`, `non-durable-exclusive`", defaultValue = "durable-exclusive")

quarkus-solace-messaging-connector/runtime/src/main/java/com/solace/quarkus/messaging/incoming/SolaceIncomingChannel.java

@@ -4,9 +4,7 @@
 
 import java.time.Duration;
 import java.time.ZonedDateTime;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
+import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Flow;
@@ -22,13 +20,16 @@
 import com.solace.messaging.config.MissingResourcesCreationConfiguration.MissingResourcesCreationStrategy;
 import com.solace.messaging.config.ReceiverActivationPassivationConfiguration;
 import com.solace.messaging.config.ReplayStrategy;
+import com.solace.messaging.config.SolaceConstants;
 import com.solace.messaging.receiver.InboundMessage;
 import com.solace.messaging.receiver.PersistentMessageReceiver;
 import com.solace.messaging.resources.Queue;
 import com.solace.messaging.resources.TopicSubscription;
 import com.solace.quarkus.messaging.SolaceConnectorIncomingConfiguration;
 import com.solace.quarkus.messaging.fault.*;
 import com.solace.quarkus.messaging.i18n.SolaceLogging;
+import com.solace.quarkus.messaging.tracing.SolaceOpenTelemetryInstrumenter;
+import com.solace.quarkus.messaging.tracing.SolaceTrace;
 
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
@@ -44,13 +45,14 @@ public class SolaceIncomingChannel implements ReceiverActivationPassivationConfi
     private final SolaceAckHandler ackHandler;
     private final SolaceFailureHandler failureHandler;
     private final AtomicBoolean closed = new AtomicBoolean(false);
-    private final AtomicBoolean alive = new AtomicBoolean(false);
+    private final AtomicBoolean alive = new AtomicBoolean(true);
     private final PersistentMessageReceiver receiver;
     private final Flow.Publisher<? extends Message<?>> stream;
     private final ExecutorService pollerThread;
     private final boolean gracefulShutdown;
     private final long gracefulShutdownWaitTimeout;
     private final List<Throwable> failures = new ArrayList<>();
+    private final SolaceOpenTelemetryInstrumenter solaceOpenTelemetryInstrumenter;
     private volatile MessagingService solace;
 
     // Assuming we won't ever exceed the limit of an unsigned long...
@@ -107,19 +109,55 @@ public SolaceIncomingChannel(Vertx vertx, SolaceConnectorIncomingConfiguration i
 
         // TODO Here use a subscription receiver.receiveAsync with an internal queue
         this.pollerThread = Executors.newSingleThreadExecutor();
-        this.stream = Multi.createBy().repeating()
+
+        Multi<? extends Message<?>> incomingMulti = Multi.createBy().repeating()
                 .uni(() -> Uni.createFrom().item(receiver::receiveMessage)
                         .runSubscriptionOn(pollerThread))
                 .until(__ -> closed.get())
                 .emitOn(context::runOnContext)
                 .map(consumed -> new SolaceInboundMessage<>(consumed, ackHandler, failureHandler,
-                        unacknowledgedMessageTracker, this::reportFailure))
-                .plug(m -> lazyStart
-                        ? m.onSubscription()
-                                .call(() -> Uni.createFrom().completionStage(receiver.startAsync()))
-                        : m)
+                        unacknowledgedMessageTracker, this::reportFailure));
+
+        if (ic.getClientTracingEnabled()) {
+            solaceOpenTelemetryInstrumenter = SolaceOpenTelemetryInstrumenter.createForIncoming();
+            incomingMulti = incomingMulti.map(message -> {
+                InboundMessage consumedMessage = message.getMetadata(SolaceInboundMetadata.class).get().getMessage();
+                Map<String, String> messageProperties = new HashMap<>();
+
+                messageProperties.put("messaging.solace.replication_group_message_id",
+                        consumedMessage.getReplicationGroupMessageId().toString());
+                messageProperties.put("messaging.solace.priority", Integer.toString(consumedMessage.getPriority()));
+                if (consumedMessage.getProperties().size() > 0) {
+                    messageProperties.putAll(consumedMessage.getProperties());
+                }
+                SolaceTrace solaceTrace = new SolaceTrace.Builder()
+                        .withDestinationKind("queue")
+                        .withTopic(consumedMessage.getDestinationName())
+                        .withMessageID(consumedMessage.getApplicationMessageId())
+                        .withCorrelationID(consumedMessage.getCorrelationId())
+                        .withPartitionKey(
+                                consumedMessage
+                                        .hasProperty(SolaceConstants.MessageUserPropertyConstants.QUEUE_PARTITION_KEY)
+                                                ? consumedMessage
+                                                        .getProperty(
+                                                                SolaceConstants.MessageUserPropertyConstants.QUEUE_PARTITION_KEY)
+                                                : null)
+                        .withPayloadSize(Long.valueOf(consumedMessage.getPayloadAsBytes().length))
+                        .withProperties(messageProperties)
+                        .build();
+                return solaceOpenTelemetryInstrumenter.traceIncoming(message, solaceTrace, true);
+            });
+        } else {
+            solaceOpenTelemetryInstrumenter = null;
+        }
+
+        this.stream = incomingMulti.plug(m -> lazyStart
+                ? m.onSubscription()
+                        .call(() -> Uni.createFrom().completionStage(receiver.startAsync()))
+                : m)
                 .onItem().invoke(() -> alive.set(true))
                 .onFailure().retry().withBackOff(Duration.ofSeconds(1)).atMost(3).onFailure().invoke(this::reportFailure);
+
         if (!lazyStart) {
             receiver.start();
         }