port;
+
+ /**
+ * The image to use.
+ * Note that only Apache Pulsar images are supported.
+ * Specifically, the image repository must end with {@code apachepulsar/pulsar}.
+ *
+ * Check https://hub.docker.com/r/apachepulsar/pulsar to find the available versions.
+ */
+ @ConfigItem(defaultValue = "apachepulsar/pulsar:3.0.0")
+ public String imageName;
+
+ /**
+ * Indicates if the Pulsar broker managed by Quarkus Dev Services is shared.
+ * When shared, Quarkus looks for running containers using label-based service discovery.
+ * If a matching container is found, it is used, and so a second one is not started.
+ * Otherwise, Dev Services for Pulsar starts a new container.
+ *
+ * The discovery uses the {@code quarkus-dev-service-pulsar} label.
+ * The value is configured using the {@code service-name} property.
+ *
+ * Container sharing is only used in dev mode.
+ */
+ @ConfigItem(defaultValue = "true")
+ public boolean shared;
+
+ /**
+ * The value of the {@code quarkus-dev-service-pulsar} label attached to the started container.
+ * This property is used when {@code shared} is set to {@code true}.
+ * In this case, before starting a container, Dev Services for Pulsar looks for a container with the
+ * {@code quarkus-dev-service-pulsar} label
+ * set to the configured value. If found, it will use this container instead of starting a new one. Otherwise, it
+ * starts a new container with the {@code quarkus-dev-service-pulsar} label set to the specified value.
+ *
+ * This property is used when you need multiple shared Pulsar brokers.
+ */
+ @ConfigItem(defaultValue = "pulsar")
+ public String serviceName;
+
+ /**
+ * Broker config to set on the Pulsar instance
+ */
+ @ConfigItem
+ public Map brokerConfig;
+
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java
new file mode 100644
index 0000000000000..6e4ff1fa51a45
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarDevServicesProcessor.java
@@ -0,0 +1,272 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar.deployment;
+
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import org.eclipse.microprofile.config.Config;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+import org.testcontainers.containers.Network;
+import org.testcontainers.utility.DockerImageName;
+
+import io.quarkus.deployment.Feature;
+import io.quarkus.deployment.IsNormal;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.BuildSteps;
+import io.quarkus.deployment.builditem.CuratedApplicationShutdownBuildItem;
+import io.quarkus.deployment.builditem.DevServicesResultBuildItem;
+import io.quarkus.deployment.builditem.DevServicesResultBuildItem.RunningDevService;
+import io.quarkus.deployment.builditem.DockerStatusBuildItem;
+import io.quarkus.deployment.builditem.LaunchModeBuildItem;
+import io.quarkus.deployment.console.ConsoleInstalledBuildItem;
+import io.quarkus.deployment.console.StartupLogCompressor;
+import io.quarkus.deployment.dev.devservices.GlobalDevServicesConfig;
+import io.quarkus.deployment.logging.LoggingSetupBuildItem;
+import io.quarkus.devservices.common.ContainerLocator;
+import io.quarkus.runtime.LaunchMode;
+import io.quarkus.runtime.configuration.ConfigUtils;
+
+/**
+ * Starts a Pulsar broker as dev service if needed.
+ * It uses https://hub.docker.com/r/apachepulsar/pulsar as image.
+ */
+@BuildSteps(onlyIfNot = IsNormal.class, onlyIf = GlobalDevServicesConfig.Enabled.class)
+public class PulsarDevServicesProcessor {
+
+ private static final Logger log = Logger.getLogger(PulsarDevServicesProcessor.class);
+
+ /**
+ * Label to add to shared Dev Service for pulsar running in containers.
+ * This allows other applications to discover the running service and use it instead of starting a new instance.
+ */
+ private static final String DEV_SERVICE_LABEL = "quarkus-dev-service-pulsar";
+
+ private static final ContainerLocator pulsarContainerLocator = new ContainerLocator(DEV_SERVICE_LABEL,
+ PulsarContainer.BROKER_PORT);
+ private static final String PULSAR_CLIENT_SERVICE_URL = "pulsar.client.serviceUrl";
+ private static final String PULSAR_ADMIN_SERVICE_URL = "pulsar.admin.serviceUrl";
+ static volatile RunningDevService devService;
+ static volatile PulsarDevServiceCfg cfg;
+ static volatile boolean first = true;
+
+ @BuildStep
+ public DevServicesResultBuildItem startPulsarDevService(
+ DockerStatusBuildItem dockerStatusBuildItem,
+ LaunchModeBuildItem launchMode,
+ PulsarBuildTimeConfig pulsarClientBuildTimeConfig,
+ Optional consoleInstalledBuildItem,
+ CuratedApplicationShutdownBuildItem closeBuildItem,
+ LoggingSetupBuildItem loggingSetupBuildItem,
+ GlobalDevServicesConfig devServicesConfig) {
+
+ PulsarDevServiceCfg configuration = getConfiguration(pulsarClientBuildTimeConfig);
+
+ if (devService != null) {
+ boolean shouldShutdownTheBroker = !configuration.equals(cfg);
+ if (!shouldShutdownTheBroker) {
+ return devService.toBuildItem();
+ }
+ shutdownBroker();
+ cfg = null;
+ }
+
+ StartupLogCompressor compressor = new StartupLogCompressor(
+ (launchMode.isTest() ? "(test) " : "") + "Pulsar Dev Services Starting:", consoleInstalledBuildItem,
+ loggingSetupBuildItem);
+ try {
+ RunningDevService newDevService = startPulsarContainer(dockerStatusBuildItem, configuration, launchMode,
+ devServicesConfig.timeout);
+ if (newDevService != null) {
+ devService = newDevService;
+ Map config = devService.getConfig();
+ if (newDevService.isOwner()) {
+ log.info("Dev Services for Pulsar started.");
+ log.infof("Other Quarkus applications in dev mode will find the "
+ + "broker automatically. For Quarkus applications in production mode, you can connect to"
+ + " this by starting your application with -Dpulsar.client.serviceUrl=%s",
+ config.get(PULSAR_CLIENT_SERVICE_URL));
+ }
+ }
+ if (devService == null) {
+ compressor.closeAndDumpCaptured();
+ } else {
+ compressor.close();
+ }
+ } catch (Throwable t) {
+ compressor.closeAndDumpCaptured();
+ throw new RuntimeException(t);
+ }
+
+ if (devService == null) {
+ return null;
+ }
+
+ // Configure the watch dog
+ if (first) {
+ first = false;
+ Runnable closeTask = () -> {
+ if (devService != null) {
+ shutdownBroker();
+
+ log.info("Dev Services for Pulsar shut down.");
+ }
+ first = true;
+ devService = null;
+ cfg = null;
+ };
+ closeBuildItem.addCloseTask(closeTask, true);
+ }
+ cfg = configuration;
+ return devService.toBuildItem();
+ }
+
+ private void shutdownBroker() {
+ if (devService != null) {
+ try {
+ devService.close();
+ } catch (Throwable e) {
+ log.error("Failed to stop the Pulsar broker", e);
+ } finally {
+ devService = null;
+ }
+ }
+ }
+
+ private RunningDevService startPulsarContainer(DockerStatusBuildItem dockerStatusBuildItem, PulsarDevServiceCfg config,
+ LaunchModeBuildItem launchMode, Optional timeout) {
+ if (!config.devServicesEnabled) {
+ // explicitly disabled
+ log.debug("Not starting Dev Services for Pulsar, as it has been disabled in the config.");
+ return null;
+ }
+
+ // Check if pulsar.serviceUrl is set
+ if (ConfigUtils.isPropertyPresent(PULSAR_CLIENT_SERVICE_URL)) {
+ log.debug("Not starting Dev Services for Pulsar, the pulsar.serviceUrl is configured.");
+ return null;
+ }
+
+ // Verify that we have Pulsar channels without host and port
+ if (!hasPulsarChannelWithoutHostAndPort()) {
+ log.debug("Not starting Dev Services for Pulsar, all the channels are configured.");
+ return null;
+ }
+
+ if (!dockerStatusBuildItem.isDockerAvailable()) {
+ log.warn("Docker isn't working, please configure the Pulsar broker location.");
+ return null;
+ }
+
+ final Supplier defaultPulsarBrokerSupplier = () -> {
+ // Starting the broker
+ PulsarContainer container = new PulsarContainer(DockerImageName.parse(config.imageName)
+ .asCompatibleSubstituteFor("apachepulsar/pulsar"))
+ .withNetwork(Network.SHARED);
+ config.brokerConfig.forEach((key, value) -> container.addEnv("PULSAR_PREFIX_" + key, value));
+ if (launchMode.getLaunchMode() == LaunchMode.DEVELOPMENT) { // Only adds the label in dev mode.
+ container.withLabel(DEV_SERVICE_LABEL, config.serviceName);
+ }
+ if (config.fixedExposedPort != 0) {
+ container.withPort(config.fixedExposedPort);
+ }
+ timeout.ifPresent(container::withStartupTimeout);
+ container.start();
+
+ return getRunningService(container.getContainerId(), container::close, container.getPulsarBrokerUrl(),
+ container.getHttpServiceUrl());
+ };
+
+ return pulsarContainerLocator.locateContainer(config.serviceName, config.shared, launchMode.getLaunchMode())
+ .map(containerAddress -> getRunningService(containerAddress.getId(), null,
+ getServiceUrl(containerAddress.getHost(), containerAddress.getPort()),
+ getHttpServiceUrl(containerAddress.getHost(),
+ pulsarContainerLocator.locatePublicPort(config.serviceName, config.shared,
+ launchMode.getLaunchMode(), PulsarContainer.BROKER_HTTP_PORT).orElse(8080))))
+ .orElseGet(defaultPulsarBrokerSupplier);
+ }
+
+ private String getServiceUrl(String host, int port) {
+ return String.format("pulsar://%s:%d", host, port);
+ }
+
+ private String getHttpServiceUrl(String host, int port) {
+ return String.format("http://%s:%d", host, port);
+ }
+
+ private RunningDevService getRunningService(String containerId, Closeable closeable, String pulsarBrokerUrl,
+ String httpServiceUrl) {
+ Map configMap = new HashMap<>();
+ configMap.put(PULSAR_CLIENT_SERVICE_URL, pulsarBrokerUrl);
+ configMap.put(PULSAR_ADMIN_SERVICE_URL, httpServiceUrl);
+ return new RunningDevService(Feature.SMALLRYE_REACTIVE_MESSAGING_PULSAR.getName(), containerId, closeable, configMap);
+ }
+
+ private boolean hasPulsarChannelWithoutHostAndPort() {
+ Config config = ConfigProvider.getConfig();
+ for (String name : config.getPropertyNames()) {
+ boolean isIncoming = name.startsWith("mp.messaging.incoming.");
+ boolean isOutgoing = name.startsWith("mp.messaging.outgoing.");
+ boolean isConnector = name.endsWith(".connector");
+ boolean isConfigured = false;
+ if ((isIncoming || isOutgoing) && isConnector) {
+ String connectorValue = config.getValue(name, String.class);
+ boolean isPulsar = connectorValue.equalsIgnoreCase("smallrye-pulsar");
+ boolean hasServiceUrl = ConfigUtils.isPropertyPresent(name.replace(".connector", ".serviceUrl"));
+ isConfigured = isPulsar && hasServiceUrl;
+ }
+
+ if (!isConfigured) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private PulsarDevServiceCfg getConfiguration(PulsarBuildTimeConfig cfg) {
+ PulsarDevServicesBuildTimeConfig devServicesConfig = cfg.devservices;
+ return new PulsarDevServiceCfg(devServicesConfig);
+ }
+
+ private static final class PulsarDevServiceCfg {
+ private final boolean devServicesEnabled;
+ private final String imageName;
+ private final Integer fixedExposedPort;
+ private final boolean shared;
+ private final String serviceName;
+ private final Map brokerConfig;
+
+ public PulsarDevServiceCfg(PulsarDevServicesBuildTimeConfig devServicesConfig) {
+ this.devServicesEnabled = devServicesConfig.enabled.orElse(true);
+ this.imageName = devServicesConfig.imageName;
+ this.fixedExposedPort = devServicesConfig.port.orElse(0);
+ this.shared = devServicesConfig.shared;
+ this.serviceName = devServicesConfig.serviceName;
+ this.brokerConfig = devServicesConfig.brokerConfig;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PulsarDevServiceCfg that = (PulsarDevServiceCfg) o;
+ return devServicesEnabled == that.devServicesEnabled && Objects.equals(imageName, that.imageName)
+ && Objects.equals(fixedExposedPort, that.fixedExposedPort)
+ && Objects.equals(brokerConfig, that.brokerConfig);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(devServicesEnabled, imageName, fixedExposedPort, brokerConfig);
+ }
+ }
+
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarSchemaDiscoveryProcessor.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarSchemaDiscoveryProcessor.java
new file mode 100644
index 0000000000000..d816493bc180e
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/PulsarSchemaDiscoveryProcessor.java
@@ -0,0 +1,584 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar.deployment;
+
+import static io.quarkus.smallrye.reactivemessaging.deployment.SmallRyeReactiveMessagingProcessor.getChannelPropertyKey;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+
+import org.apache.pulsar.common.schema.SchemaType;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+import org.jboss.jandex.MethodParameterInfo;
+import org.jboss.jandex.Type;
+import org.jboss.logging.Logger;
+
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.processor.KotlinUtils;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.ExecutionTime;
+import io.quarkus.deployment.annotations.Record;
+import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
+import io.quarkus.deployment.builditem.RunTimeConfigurationDefaultBuildItem;
+import io.quarkus.deployment.recording.RecorderContext;
+import io.quarkus.pulsar.SchemaProviderRecorder;
+import io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames;
+import io.quarkus.smallrye.reactivemessaging.deployment.items.ConnectorManagedChannelBuildItem;
+
+public class PulsarSchemaDiscoveryProcessor {
+
+ static Logger log = Logger.getLogger(PulsarSchemaDiscoveryProcessor.class);
+
+ /**
+ * Handles the serializer/deserializer detection and whether the graceful shutdown should be used in dev mode.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ public void defaultChannelConfiguration(
+ ReactiveMessagingPulsarBuildTimeConfig buildTimeConfig,
+ CombinedIndexBuildItem combinedIndex,
+ List channelsManagedByConnectors,
+ BuildProducer defaultConfigProducer,
+ BuildProducer syntheticBean,
+ RecorderContext recorderContext,
+ SchemaProviderRecorder recorder) {
+ if (buildTimeConfig.schemaAutodetectionEnabled) {
+ DefaultSchemaDiscoveryState discoveryState = new DefaultSchemaDiscoveryState(combinedIndex.getIndex());
+ discoverDefaultSerdeConfig(discoveryState, channelsManagedByConnectors, defaultConfigProducer,
+ buildTimeConfig.schemaGenerationEnabled ? new SyntheticBeanBuilder(syntheticBean, recorder, recorderContext)
+ : null);
+ }
+ }
+
+ // visible for testing
+ void discoverDefaultSerdeConfig(DefaultSchemaDiscoveryState discovery,
+ List channelsManagedByConnectors,
+ BuildProducer config,
+ SyntheticBeanBuilder syntheticBean) {
+ for (AnnotationInstance annotation : discovery.findRepeatableAnnotationsOnMethods(DotNames.INCOMING)) {
+ String channelName = annotation.value().asString();
+ if (!discovery.isPulsarConnector(channelsManagedByConnectors, true, channelName)) {
+ continue;
+ }
+
+ MethodInfo method = annotation.target().asMethod();
+
+ Type incomingType = getIncomingTypeFromMethod(method);
+ processIncomingType(discovery, config, incomingType, channelName, syntheticBean);
+ }
+
+ for (AnnotationInstance annotation : discovery.findAnnotationsOnMethods(DotNames.OUTGOING)) {
+ String channelName = annotation.value().asString();
+ if (!discovery.isPulsarConnector(channelsManagedByConnectors, false, channelName)) {
+ continue;
+ }
+
+ MethodInfo method = annotation.target().asMethod();
+
+ Type outgoingType = getOutgoingTypeFromMethod(method);
+ processOutgoingType(discovery, config, outgoingType, channelName, syntheticBean);
+ }
+
+ for (AnnotationInstance annotation : discovery.findAnnotationsOnInjectionPoints(DotNames.CHANNEL)) {
+ String channelName = annotation.value().asString();
+ if (!discovery.isPulsarConnector(channelsManagedByConnectors, false, channelName)
+ && !discovery.isPulsarConnector(channelsManagedByConnectors, true, channelName)) {
+ continue;
+ }
+
+ Type injectionPointType = getInjectionPointType(annotation);
+ if (injectionPointType == null) {
+ continue;
+ }
+
+ Type incomingType = getIncomingTypeFromChannelInjectionPoint(injectionPointType);
+ processIncomingType(discovery, config, incomingType, channelName, syntheticBean);
+ processPulsarTransactions(discovery, config, channelName, injectionPointType);
+
+ Type outgoingType = getOutgoingTypeFromChannelInjectionPoint(injectionPointType);
+ processOutgoingType(discovery, config, outgoingType, channelName, syntheticBean);
+ }
+ }
+
+ private static String outgoingSchemaKey(String channelName) {
+ return getChannelPropertyKey(channelName, "schema", false);
+ }
+
+ private void processPulsarTransactions(DefaultSchemaDiscoveryState discovery,
+ BuildProducer config,
+ String channelName,
+ Type injectionPointType) {
+ if (injectionPointType != null && isPulsarEmitter(injectionPointType)) {
+ String enableTransactionKey = getChannelPropertyKey(channelName, "enableTransaction", false);
+ log.infof("Transactional producer detected for channel '%s', setting following default config values: "
+ + "'" + enableTransactionKey + "=true'", channelName);
+ produceRuntimeConfigurationDefaultBuildItem(discovery, config, enableTransactionKey, "true");
+ }
+ }
+
+ private void processIncomingType(DefaultSchemaDiscoveryState discovery,
+ BuildProducer config,
+ Type incomingType,
+ String channelName,
+ SyntheticBeanBuilder syntheticBean) {
+ extractValueType(incomingType, (value, isBatch) -> {
+ if (discovery.findProvidedSchemaWithIdentifier(channelName).isEmpty()) {
+ if (discovery.hasObjectMapperConfigSchema(value, channelName, true)) {
+ objectMapperSchemaFor(SyntheticBeanBuilder.objectMapperSchemaId(value), value, syntheticBean);
+ } else {
+ String schema = schemaFor(discovery, value, syntheticBean);
+ produceRuntimeConfigurationDefaultBuildItem(discovery, config, incomingSchemaKey(channelName), schema);
+ }
+ }
+ if (Boolean.TRUE.equals(isBatch)) {
+ produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+ getChannelPropertyKey(channelName, "batchReceive", true), "true");
+ }
+ });
+ }
+
+ private static String incomingSchemaKey(String channelName) {
+ return getChannelPropertyKey(channelName, "schema", true);
+ }
+
+ private Type getInjectionPointType(AnnotationInstance annotation) {
+ switch (annotation.target().kind()) {
+ case FIELD:
+ return annotation.target().asField().type();
+ case METHOD_PARAMETER:
+ MethodParameterInfo parameter = annotation.target().asMethodParameter();
+ return parameter.method().parameterType(parameter.position());
+ default:
+ return null;
+ }
+ }
+
+ private void produceRuntimeConfigurationDefaultBuildItem(DefaultSchemaDiscoveryState discovery,
+ BuildProducer config,
+ String key, String value) {
+ if (value == null) {
+ return;
+ }
+
+ if (discovery.shouldNotConfigure(key)) {
+ return;
+ }
+
+ discovery.ifNotYetConfigured(key, () -> config.produce(new RunTimeConfigurationDefaultBuildItem(key, value)));
+ }
+
+ private Type getIncomingTypeFromMethod(MethodInfo method) {
+ List parameterTypes = method.parameterTypes();
+ int parametersCount = parameterTypes.size();
+ Type returnType = method.returnType();
+
+ Type incomingType = null;
+
+ // @Incoming
+ if ((isVoid(returnType) && parametersCount >= 1)
+ || (isCompletionStage(returnType) && parametersCount >= 1)
+ || (isUni(returnType) && parametersCount >= 1)) {
+ incomingType = parameterTypes.get(0);
+ } else if ((isSubscriber(returnType) && parametersCount == 0)
+ || (isSubscriberBuilder(returnType) && parametersCount == 0)) {
+ incomingType = returnType.asParameterizedType().arguments().get(0);
+ } else if (KotlinUtils.isKotlinSuspendMethod(method)) {
+ incomingType = parameterTypes.get(0);
+ }
+
+ // @Incoming @Outgoing
+ if (method.hasAnnotation(DotNames.OUTGOING)) {
+ if ((isCompletionStage(returnType) && parametersCount >= 1)
+ || (isUni(returnType) && parametersCount >= 1)
+ || (isPublisher(returnType) && parametersCount == 1)
+ || (isPublisherBuilder(returnType) && parametersCount == 1)
+ || (isFlowPublisher(returnType) && parametersCount == 1)
+ || (isMulti(returnType) && parametersCount == 1)) {
+ incomingType = parameterTypes.get(0);
+ } else if ((isProcessor(returnType) && parametersCount == 0)
+ || (isProcessorBuilder(returnType) && parametersCount == 0)) {
+ incomingType = returnType.asParameterizedType().arguments().get(0);
+ } else if (parametersCount >= 1) {
+ incomingType = parameterTypes.get(0);
+ } else if (KotlinUtils.isKotlinSuspendMethod(method)) {
+ incomingType = parameterTypes.get(0);
+ }
+
+ // @Incoming @Outgoing stream manipulation
+ if (incomingType != null
+ && (isPublisher(incomingType) || isPublisherBuilder(incomingType)
+ || isMulti(incomingType) || isFlowPublisher(incomingType))) {
+ incomingType = incomingType.asParameterizedType().arguments().get(0);
+ }
+ }
+ return incomingType;
+ }
+
+ private Type getIncomingTypeFromChannelInjectionPoint(Type injectionPointType) {
+ if (injectionPointType == null) {
+ return null;
+ }
+
+ if (isPublisher(injectionPointType) || isPublisherBuilder(injectionPointType)
+ || isMulti(injectionPointType) || isFlowPublisher(injectionPointType)) {
+ return injectionPointType.asParameterizedType().arguments().get(0);
+ } else {
+ return null;
+ }
+ }
+
+ private Type getOutgoingTypeFromMethod(MethodInfo method) {
+ List parameterTypes = method.parameterTypes();
+ int parametersCount = parameterTypes.size();
+ Type returnType = method.returnType();
+
+ Type outgoingType = null;
+
+ // @Outgoing
+ if ((isPublisher(returnType) && parametersCount == 0)
+ || (isPublisherBuilder(returnType) && parametersCount == 0)
+ || (isMulti(returnType) && parametersCount == 0)
+ || (isFlowPublisher(returnType) && parametersCount == 0)
+ || (isCompletionStage(returnType) && parametersCount == 0)
+ || (isUni(returnType) && parametersCount == 0)) {
+ outgoingType = returnType.asParameterizedType().arguments().get(0);
+ } else if (parametersCount == 0) {
+ outgoingType = returnType;
+ } else if (KotlinUtils.isKotlinSuspendMethod(method)) {
+ outgoingType = getReturnTypeFromKotlinSuspendMethod(method);
+ }
+
+ // @Incoming @Outgoing
+ if (method.hasAnnotation(DotNames.INCOMING)) {
+ if ((isCompletionStage(returnType) && parametersCount == 1)
+ || (isUni(returnType) && parametersCount == 1)
+ || (isPublisher(returnType) && parametersCount == 1)
+ || (isPublisherBuilder(returnType) && parametersCount == 1)
+ || (isFlowPublisher(returnType) && parametersCount == 1)
+ || (isMulti(returnType) && parametersCount == 1)) {
+ outgoingType = returnType.asParameterizedType().arguments().get(0);
+ } else if ((isProcessor(returnType) && parametersCount == 0)
+ || (isProcessorBuilder(returnType) && parametersCount == 0)) {
+ outgoingType = returnType.asParameterizedType().arguments().get(1);
+ } else if (parametersCount == 1) {
+ outgoingType = returnType;
+ } else if (KotlinUtils.isKotlinSuspendMethod(method)) {
+ outgoingType = getReturnTypeFromKotlinSuspendMethod(method);
+ }
+
+ // @Incoming @Outgoing stream manipulation
+ if (outgoingType != null
+ && (isPublisher(outgoingType) || isPublisherBuilder(outgoingType)
+ || isMulti(outgoingType) || isFlowPublisher(outgoingType))) {
+ outgoingType = outgoingType.asParameterizedType().arguments().get(0);
+ }
+ }
+ return outgoingType;
+ }
+
+ private static Type getReturnTypeFromKotlinSuspendMethod(MethodInfo method) {
+ Type continuationReturnType = method.parameterType(method.parametersCount() - 1);
+
+ if (continuationReturnType.kind() == Type.Kind.PARAMETERIZED_TYPE) {
+ Type firstGenericType = continuationReturnType.asParameterizedType().arguments().get(0);
+ if (firstGenericType.kind() == Type.Kind.WILDCARD_TYPE) {
+ return firstGenericType.asWildcardType().superBound();
+ }
+ }
+
+ return null;
+ }
+
+ private Type getOutgoingTypeFromChannelInjectionPoint(Type injectionPointType) {
+ if (injectionPointType == null) {
+ return null;
+ }
+
+ if (isEmitter(injectionPointType) || isMutinyEmitter(injectionPointType) || isPulsarEmitter(injectionPointType)) {
+ return injectionPointType.asParameterizedType().arguments().get(0);
+ } else {
+ return null;
+ }
+ }
+
+ private void processOutgoingType(DefaultSchemaDiscoveryState discovery,
+ BuildProducer config,
+ Type outgoingType,
+ String channelName, SyntheticBeanBuilder syntheticBean) {
+ extractValueType(outgoingType, (value, isBatch) -> {
+ if (discovery.findProvidedSchemaWithIdentifier(channelName).isEmpty()) {
+ if (discovery.hasObjectMapperConfigSchema(value, channelName, false)) {
+ objectMapperSchemaFor(SyntheticBeanBuilder.objectMapperSchemaId(value), value, syntheticBean);
+ } else {
+ String schema = schemaFor(discovery, value, syntheticBean);
+ produceRuntimeConfigurationDefaultBuildItem(discovery, config, outgoingSchemaKey(channelName), schema);
+ }
+ }
+ });
+ }
+
+ private void extractValueType(Type type, BiConsumer schemaAcceptor) {
+ if (type == null) {
+ return;
+ }
+
+ if (isMessage(type)) {
+ List typeArguments = type.asParameterizedType().arguments();
+ Type messageTypeParameter = typeArguments.get(0);
+ if (isList(messageTypeParameter)) {
+ List messageListTypeArguments = messageTypeParameter.asParameterizedType().arguments();
+ schemaAcceptor.accept(messageListTypeArguments.get(0), true);
+ } else {
+ schemaAcceptor.accept(messageTypeParameter, false);
+ }
+ } else if (isList(type)) {
+ List typeArguments = type.asParameterizedType().arguments();
+ schemaAcceptor.accept(typeArguments.get(0), true);
+ } else if (isPulsarMessage(type) || isPulsarApiMessage(type) || isOutgoingMessage(type)) {
+ List typeArguments = type.asParameterizedType().arguments();
+ schemaAcceptor.accept(typeArguments.get(0), false);
+ } else if (isPulsarApiMessages(type) || isPulsarBatchMessage(type)) {
+ List typeArguments = type.asParameterizedType().arguments();
+ schemaAcceptor.accept(typeArguments.get(0), true);
+ } else if (isKeyedMulti(type)) {
+ List typeArguments = type.asParameterizedType().arguments();
+ schemaAcceptor.accept(typeArguments.get(1), false);
+ } else if (isRawMessage(type)) {
+ schemaAcceptor.accept(type, false);
+ }
+ }
+
+ // ---
+
+ private static boolean isVoid(Type type) {
+ return type.kind() == Type.Kind.VOID;
+ }
+
+ private static boolean isCompletionStage(Type type) {
+ // raw type CompletionStage is wrong, must be CompletionStage
+ return DotNames.COMPLETION_STAGE.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isUni(Type type) {
+ // raw type Uni is wrong, must be Uni
+ return DotNames.UNI.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isMulti(Type type) {
+ // raw type Multi is wrong, must be Multi
+ return DotNames.MULTI.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isFlowPublisher(Type type) {
+ // raw type Flow.Publisher is wrong, must be Multi
+ return DotNames.FLOW_PUBLISHER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isSubscriber(Type type) {
+ // raw type Subscriber is wrong, must be Subscriber
+ return DotNames.SUBSCRIBER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isSubscriberBuilder(Type type) {
+ // raw type SubscriberBuilder is wrong, must be SubscriberBuilder
+ return DotNames.SUBSCRIBER_BUILDER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 2;
+ }
+
+ private static boolean isPublisher(Type type) {
+ // raw type Publisher is wrong, must be Publisher
+ return DotNames.PUBLISHER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isPublisherBuilder(Type type) {
+ // raw type PublisherBuilder is wrong, must be PublisherBuilder
+ return DotNames.PUBLISHER_BUILDER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isProcessor(Type type) {
+ // raw type Processor is wrong, must be Processor
+ return DotNames.PROCESSOR.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 2;
+ }
+
+ private static boolean isProcessorBuilder(Type type) {
+ // raw type ProcessorBuilder is wrong, must be ProcessorBuilder
+ return DotNames.PROCESSOR_BUILDER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 2;
+ }
+
+ // ---
+
+ private static boolean isEmitter(Type type) {
+ // raw type Emitter is wrong, must be Emitter
+ return DotNames.EMITTER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isMutinyEmitter(Type type) {
+ // raw type MutinyEmitter is wrong, must be MutinyEmitter
+ return DotNames.MUTINY_EMITTER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isPulsarEmitter(Type type) {
+ // raw type PulsarTransactions is wrong, must be PulsarTransactions
+ return DotNames.PULSAR_EMITTER.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ // ---
+
+ private static boolean isMessage(Type type) {
+ // raw type Message is wrong, must be Message
+ return DotNames.MESSAGE.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isPulsarMessage(Type type) {
+ // raw type PulsarMessage is wrong, must be PulsarMessage
+ return DotNames.PULSAR_MESSAGE.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isOutgoingMessage(Type type) {
+ // raw type Record is wrong, must be Record
+ return DotNames.OUTGOING_MESSAGE.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isPulsarApiMessage(Type type) {
+ // raw type ConsumerRecord is wrong, must be ConsumerRecord
+ return DotNames.PULSAR_API_MESSAGE.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isKeyedMulti(Type type) {
+ return ReactiveMessagingDotNames.KEYED_MULTI.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 2;
+ }
+
+ private static boolean isList(Type type) {
+ return DotNames.LIST.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isPulsarBatchMessage(Type type) {
+ return DotNames.PULSAR_BATCH_MESSAGE.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isPulsarApiMessages(Type type) {
+ return DotNames.PULSAR_API_MESSAGES.equals(type.name())
+ && type.kind() == Type.Kind.PARAMETERIZED_TYPE
+ && type.asParameterizedType().arguments().size() == 1;
+ }
+
+ private static boolean isRawMessage(Type type) {
+ switch (type.kind()) {
+ case PRIMITIVE:
+ case CLASS:
+ case ARRAY:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ private String knownSchemaFor(Type type) {
+ if (type == null) {
+ return null;
+ }
+ DotName typeName = type.name();
+
+ // statically known schema
+ if (KNOWN_SCHEMA.containsKey(typeName)) {
+ return KNOWN_SCHEMA.get(typeName);
+ }
+
+ // unknown
+ return null;
+ }
+
+ private void objectMapperSchemaFor(String schemaId, Type type, SyntheticBeanBuilder syntheticBean) {
+ if (syntheticBean != null) {
+ syntheticBean.produceObjectMapperSchemaBean(schemaId, type);
+ }
+ }
+
+ private String schemaFor(DefaultSchemaDiscoveryState discovery, Type type, SyntheticBeanBuilder syntheticBean) {
+ String result = knownSchemaFor(type);
+ if (result == null && type != null && syntheticBean != null && type.kind() == Type.Kind.CLASS) {
+ String schemaId = syntheticBean.produceSchemaBean(discovery, type);
+ if (schemaId != null) {
+ log.infof("Generating Schema for type %s %s", type.name().toString(), schemaId);
+ }
+ result = schemaId;
+ }
+ return result;
+ }
+
+ // ---
+
+ // @formatter:off
+ private static final Map KNOWN_SCHEMA = Map.ofEntries(
+ // primitives
+ Map.entry(DotName.createSimple("byte"), SchemaType.INT8.name()),
+ Map.entry(DotName.createSimple("short"), SchemaType.INT16.name()),
+ Map.entry(DotName.createSimple("int"), SchemaType.INT32.name()),
+ Map.entry(DotName.createSimple("long"), SchemaType.INT64.name()),
+ Map.entry(DotName.createSimple("float"), SchemaType.FLOAT.name()),
+ Map.entry(DotName.createSimple("double"), SchemaType.DOUBLE.name()),
+ Map.entry(DotName.createSimple("boolean"), SchemaType.BOOLEAN.name()),
+ // primitive wrappers
+ Map.entry(DotName.createSimple(Byte.class.getName()), SchemaType.INT8.name()),
+ Map.entry(DotName.createSimple(Short.class.getName()), SchemaType.INT16.name()),
+ Map.entry(DotName.createSimple(Integer.class.getName()), SchemaType.INT32.name()),
+ Map.entry(DotName.createSimple(Long.class.getName()), SchemaType.INT64.name()),
+ Map.entry(DotName.createSimple(Float.class.getName()), SchemaType.FLOAT.name()),
+ Map.entry(DotName.createSimple(Double.class.getName()), SchemaType.DOUBLE.name()),
+ Map.entry(DotName.createSimple(Boolean.class.getName()), SchemaType.BOOLEAN.name()),
+ // arrays
+ Map.entry(DotName.createSimple("[B"), SchemaType.BYTES.name()),
+ // other
+ Map.entry(DotName.createSimple(String.class.getName()), SchemaType.STRING.name()),
+ Map.entry(DotName.createSimple(java.time.Instant.class.getName()), SchemaType.INSTANT.name()),
+ Map.entry(DotName.createSimple(java.sql.Timestamp.class.getName()), SchemaType.TIMESTAMP.name()),
+ Map.entry(DotName.createSimple(java.time.LocalDate.class.getName()), SchemaType.LOCAL_DATE.name()),
+ Map.entry(DotName.createSimple(java.time.LocalTime.class.getName()), SchemaType.LOCAL_TIME.name()),
+ Map.entry(DotName.createSimple(java.time.LocalDateTime.class.getName()), SchemaType.LOCAL_DATE_TIME.name()),
+ // Pulsar
+ Map.entry(DotNames.PULSAR_GENERIC_RECORD, SchemaType.AUTO_CONSUME.name())
+ );
+ // @formatter:on
+
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/ReactiveMessagingPulsarBuildTimeConfig.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/ReactiveMessagingPulsarBuildTimeConfig.java
new file mode 100644
index 0000000000000..34cde49902c8f
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/ReactiveMessagingPulsarBuildTimeConfig.java
@@ -0,0 +1,21 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar.deployment;
+
+import io.quarkus.runtime.annotations.ConfigItem;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+
+@ConfigRoot(name = "reactive-messaging.pulsar", phase = ConfigPhase.BUILD_TIME)
+public class ReactiveMessagingPulsarBuildTimeConfig {
+ /**
+ * Whether or not Pulsar Schema auto-detection is enabled.
+ */
+ @ConfigItem(name = "schema-autodetection.enabled", defaultValue = "true")
+ public boolean schemaAutodetectionEnabled;
+
+ /**
+ * Whether Pulsar Schema generation is enabled.
+ * When no Schema are found and not set, Quarkus generates a JSON Schema.
+ */
+ @ConfigItem(name = "schema-generation.enabled", defaultValue = "true")
+ public boolean schemaGenerationEnabled;
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/SmallRyeReactiveMessagingPulsarProcessor.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/SmallRyeReactiveMessagingPulsarProcessor.java
new file mode 100644
index 0000000000000..2e3096fbdcbf7
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/SmallRyeReactiveMessagingPulsarProcessor.java
@@ -0,0 +1,189 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar.deployment;
+
+import java.util.Collection;
+import java.util.function.BiFunction;
+import java.util.logging.Level;
+
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.jboss.jandex.ClassInfo;
+import org.jboss.logging.Logger;
+import org.objectweb.asm.ClassVisitor;
+import org.objectweb.asm.FieldVisitor;
+import org.objectweb.asm.MethodVisitor;
+import org.objectweb.asm.Opcodes;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.deployment.Feature;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.builditem.BytecodeTransformerBuildItem;
+import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
+import io.quarkus.deployment.builditem.ExtensionSslNativeSupportBuildItem;
+import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
+import io.quarkus.deployment.builditem.LogCategoryBuildItem;
+import io.quarkus.deployment.builditem.RunTimeConfigurationDefaultBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.NativeImageConfigBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
+import io.quarkus.gizmo.Gizmo;
+import io.quarkus.pulsar.PulsarRuntimeConfigProducer;
+
+public class SmallRyeReactiveMessagingPulsarProcessor {
+
+ static Logger log = Logger.getLogger(SmallRyeReactiveMessagingPulsarProcessor.class);
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(Feature.SMALLRYE_REACTIVE_MESSAGING_PULSAR);
+ }
+
+ @BuildStep
+ public AdditionalBeanBuildItem runtimeConfig() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClass(PulsarRuntimeConfigProducer.class)
+ .setUnremovable()
+ .build();
+ }
+
+ @BuildStep
+ void logging(BuildProducer log) {
+ log.produce(new LogCategoryBuildItem("org.apache.pulsar.common.util.netty.DnsResolverUtil", Level.OFF));
+ }
+
+ @BuildStep
+ void disableStatsLogging(BuildProducer runtimeConfig) {
+ runtimeConfig.produce(new RunTimeConfigurationDefaultBuildItem(
+ "mp.messaging.connector.smallrye-pulsar.statsIntervalSeconds", "0"));
+ }
+
+ @BuildStep
+ NativeImageResourceBuildItem nativeImageResources() {
+ return new NativeImageResourceBuildItem(
+ "org/asynchttpclient/config/ahc-default.properties",
+ "org/asynchttpclient/config/ahc.properties");
+ }
+
+ @BuildStep
+ void bytecodeTransformer(BuildProducer producer) {
+ String klass = "org.asynchttpclient.request.body.multipart.FileLikePart";
+
+ producer.produce(new BytecodeTransformerBuildItem(klass, new BiFunction() {
+ @Override
+ public ClassVisitor apply(String cls, ClassVisitor classVisitor) {
+ return new FileLikePartJavaxRemover(Gizmo.ASM_API_VERSION, classVisitor);
+ }
+
+ }));
+ }
+
+ private class FileLikePartJavaxRemover extends ClassVisitor {
+ public FileLikePartJavaxRemover(int version, ClassVisitor cv) {
+ super(version, cv);
+ log.debug("Removing javax.activation from FileLikePart");
+ }
+
+ @Override
+ public FieldVisitor visitField(int access, String name, String descriptor, String signature, Object value) {
+ if (name.equals("MIME_TYPES_FILE_TYPE_MAP")) {
+ return null;
+ }
+ return super.visitField(access, name, descriptor, signature, value);
+ }
+
+ // invoked for every method
+ @Override
+ public MethodVisitor visitMethod(int access, String name, String desc, String signature, String[] exceptions) {
+ MethodVisitor visitor = super.visitMethod(access, name, desc, signature, exceptions);
+ if (visitor == null) {
+ return null;
+ }
+ if (name.equals("")) {
+ return new MethodVisitor(Gizmo.ASM_API_VERSION, visitor) {
+ @Override
+ public void visitCode() {
+ mv.visitCode();
+ mv.visitInsn(Opcodes.RETURN);// our new code
+ }
+ };
+ }
+ if (name.equals("computeContentType")) {
+ return new MethodVisitor(Gizmo.ASM_API_VERSION, visitor) {
+ @Override
+ public void visitCode() {
+ super.visitCode();
+ visitVarInsn(Opcodes.ALOAD, 1); // first param
+ visitInsn(Opcodes.ARETURN);
+ }
+ };
+ }
+ visitor.visitMaxs(0, 0);
+ return visitor;
+ }
+
+ }
+
+ @BuildStep
+ IndexDependencyBuildItem indexPulsar() {
+ return new IndexDependencyBuildItem("org.apache.pulsar", "pulsar-client-original");
+ }
+
+ @BuildStep
+ public NativeImageConfigBuildItem pulsarRuntimeInitialized(
+ CombinedIndexBuildItem combinedIndex,
+ BuildProducer reflectiveClass,
+ BuildProducer nativeSslSupport) {
+ nativeSslSupport.produce(new ExtensionSslNativeSupportBuildItem(Feature.SMALLRYE_REACTIVE_MESSAGING_PULSAR));
+ reflectiveClass.produce(ReflectiveClassBuildItem
+ .builder(ClientConfigurationData.class.getName(),
+ ProducerConfigurationData.class.getName(),
+ ConsumerConfigurationData.class.getName(),
+ "com.google.protobuf.GeneratedMessageV3",
+ "org.apache.pulsar.common.protocol.schema.ProtobufNativeSchemaData",
+ "org.apache.pulsar.client.impl.schema.ProtobufNativeSchema$ProtoBufParsingInfo",
+ "org.apache.pulsar.client.impl.schema.ProtobufSchema$ProtoBufParsingInfo",
+ "org.apache.pulsar.common.schema.KeyValue")
+ .fields(true)
+ .methods(true)
+ .constructors(true)
+ .build());
+ reflectiveClass.produce(ReflectiveClassBuildItem
+ .builder("org.apache.pulsar.client.util.SecretsSerializer")
+ .constructors().build());
+
+ Collection authPluginClasses = combinedIndex.getIndex()
+ .getAllKnownImplementors(DotNames.PULSAR_AUTHENTICATION);
+ for (ClassInfo authPluginClass : authPluginClasses) {
+ reflectiveClass.produce(ReflectiveClassBuildItem.builder(authPluginClass.name().toString())
+ .constructors().build());
+ }
+
+ return NativeImageConfigBuildItem.builder()
+ .addNativeImageSystemProperty("io.netty.handler.ssl.noOpenSsl", "true")
+ .addRuntimeInitializedClass("org.apache.pulsar.common.allocator.PulsarByteBufAllocator")
+ .addRuntimeInitializedClass("org.apache.pulsar.common.protocol.Commands")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.impl.auth.oauth2.protocol.TokenClient")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.impl.crypto.MessageCryptoBc")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.impl.Backoff")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.impl.ConnectionPool")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.impl.ControlledClusterFailover")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.impl.HttpClient")
+ .addRuntimeInitializedClass("org.apache.pulsar.client.util.WithSNISslEngineFactory")
+ .addRuntimeInitializedClass("com.yahoo.sketches.quantiles.DoublesSketch")
+ .addRuntimeInitializedClass("io.netty.buffer.PooledByteBufAllocator")
+ .addRuntimeInitializedClass("io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledHeapByteBuf")
+ .addRuntimeInitializedClass("io.netty.incubator.channel.uring.IOUringEventLoopGroup")
+ .addRuntimeInitializedClass("io.netty.incubator.channel.uring.Native")
+ .addRuntimeInitializedClass("io.netty.incubator.channel.uring.IOUring")
+ .addRuntimeInitializedClass("org.asynchttpclient.RequestBuilderBase")
+ .addRuntimeInitializedClass("org.asynchttpclient.RequestBuilder")
+ .addRuntimeInitializedClass("org.asynchttpclient.BoundRequestBuilder")
+ .addRuntimeInitializedClass("org.asynchttpclient.ntlm.NtlmEngine")
+ .addRuntimeInitializedClass("sun.awt.dnd.SunDropTargetContextPeer$EventDispatcher")
+ .build();
+ }
+
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/SyntheticBeanBuilder.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/SyntheticBeanBuilder.java
new file mode 100644
index 0000000000000..bbce1efc73f3d
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/SyntheticBeanBuilder.java
@@ -0,0 +1,124 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar.deployment;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Supplier;
+
+import jakarta.enterprise.context.Dependent;
+
+import org.jboss.jandex.ParameterizedType;
+import org.jboss.jandex.Type;
+
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.recording.RecorderContext;
+import io.quarkus.pulsar.SchemaProviderRecorder;
+import io.quarkus.runtime.RuntimeValue;
+import io.quarkus.runtime.util.HashUtil;
+import io.smallrye.common.annotation.Identifier;
+
+public class SyntheticBeanBuilder {
+ final BuildProducer syntheticBeanBuildItem;
+ final SchemaProviderRecorder recorder;
+ final RecorderContext recorderContext;
+ final Map alreadyGeneratedSchema;
+
+ public SyntheticBeanBuilder(BuildProducer syntheticBeanBuildItem,
+ SchemaProviderRecorder recorder,
+ RecorderContext recorderContext) {
+ this.syntheticBeanBuildItem = syntheticBeanBuildItem;
+ this.recorder = recorder;
+ this.recorderContext = recorderContext;
+ this.alreadyGeneratedSchema = new HashMap<>();
+ }
+
+ static String objectMapperSchemaId(Type type) {
+ return "ObjectMapper<" + type.name() + ">";
+ }
+
+ String schemaIdFor(Type type) {
+ return alreadyGeneratedSchema.get(type.toString());
+ }
+
+ void produceObjectMapperSchemaBean(String schemaId, Type type) {
+ if (!alreadyGeneratedSchema.containsKey(type.toString())
+ || alreadyGeneratedSchema.get(type.toString()).equals(schemaId)) {
+ var runtimeValue = recorder.createObjectMapperSchema(recorderContext.classProxy(type.name().toString()));
+ produceSyntheticBeanSchema(syntheticBeanBuildItem, runtimeValue, schemaId, type);
+ alreadyGeneratedSchema.put(type.toString(), schemaId);
+ }
+ }
+
+ public String produceSchemaBean(DefaultSchemaDiscoveryState discovery, Type type) {
+ if (syntheticBeanBuildItem != null && type.kind() == Type.Kind.CLASS) {
+ String schemaId = schemaIdFor(type);
+ if (schemaId == null) {
+ if (discovery.isAvroGenerated(type.name()) || DotNames.AVRO_GENERIC_RECORD.equals(type.name())) {
+ schemaId = generateId(type, "AVRO");
+ produceSyntheticBeanSchema(syntheticBeanBuildItem,
+ recorder.createAvroSchema(recorderContext.classProxy(type.name().toString())), schemaId, type);
+ } else if (discovery.isProtobufGenerated(type.name())) {
+ schemaId = generateId(type, "PROTOBUF");
+ produceSyntheticBeanSchema(syntheticBeanBuildItem,
+ recorder.createProtoBufSchema(recorderContext.classProxy(type.name().toString())), schemaId, type);
+ } else if (type.name().equals(DotNames.VERTX_JSON_OBJECT)) {
+ schemaId = generateId(type, "JSON_OBJECT");
+ produceSyntheticBeanSchema(syntheticBeanBuildItem, recorder.createJsonObjectSchema(), schemaId, type);
+ } else if (type.name().equals(DotNames.VERTX_JSON_ARRAY)) {
+ schemaId = generateId(type, "JSON_ARRAY");
+ produceSyntheticBeanSchema(syntheticBeanBuildItem, recorder.createJsonArraySchema(), schemaId, type);
+ } else if (type.name().equals(DotNames.VERTX_BUFFER)) {
+ schemaId = generateId(type, "BUFFER");
+ produceSyntheticBeanSchema(syntheticBeanBuildItem, recorder.createBufferSchema(), schemaId, type);
+ } else if (type.name().equals(DotNames.BYTE_BUFFER)) {
+ schemaId = generateId(type, "BYTE_BUFFER");
+ produceSyntheticBeanSchema(syntheticBeanBuildItem, recorder.createByteBufferSchema(), schemaId, type);
+ } else {
+ schemaId = generateId(type, "JSON");
+ produceSyntheticBeanSchema(syntheticBeanBuildItem,
+ recorder.createJsonSchema(recorderContext.classProxy(type.name().toString())), schemaId, type);
+ }
+ alreadyGeneratedSchema.put(type.toString(), schemaId);
+ }
+ return schemaId;
+ }
+ return null;
+ }
+
+ void produceSyntheticBeanSchema(BuildProducer syntheticBeanBuildItem,
+ RuntimeValue runtimeValue,
+ String schemaId,
+ Type type) {
+ ParameterizedType providerType = ParameterizedType.create(DotNames.PULSAR_SCHEMA, type);
+ syntheticBeanBuildItem.produce(SyntheticBeanBuildItem.configure(Object.class)
+ .providerType(providerType)
+ .addType(providerType)
+ .addQualifier().annotation(Identifier.class).addValue("value", schemaId).done()
+ .scope(Dependent.class)
+ .runtimeValue(runtimeValue)
+ .unremovable()
+ .done());
+ }
+
+ void produceSyntheticBeanSchema(BuildProducer syntheticBeanBuildItem,
+ Supplier supplier,
+ String schemaId,
+ Type type) {
+ ParameterizedType providerType = ParameterizedType.create(DotNames.PULSAR_SCHEMA, type);
+ syntheticBeanBuildItem.produce(SyntheticBeanBuildItem.configure(Object.class)
+ .providerType(providerType)
+ .addType(providerType)
+ .addQualifier().annotation(Identifier.class).addValue("value", schemaId).done()
+ .scope(Dependent.class)
+ .supplier(supplier)
+ .unremovable()
+ .done());
+ }
+
+ String generateId(Type type, String targetType) {
+ String baseName = type.name().withoutPackagePrefix();
+ return baseName + "_" + targetType + "Schema_"
+ + HashUtil.sha1(Long.toString(UUID.randomUUID().getMostSignificantBits()));
+ }
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/ConsumingBean.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/ConsumingBean.java
new file mode 100644
index 0000000000000..b77a144550a63
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/ConsumingBean.java
@@ -0,0 +1,21 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+
+@ApplicationScoped
+public class ConsumingBean {
+
+ volatile long last = -1;
+
+ @Incoming("in")
+ public void consume(long content) {
+ last = content;
+ }
+
+ public long get() {
+ return last;
+ }
+
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/ProducingBean.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/ProducingBean.java
new file mode 100644
index 0000000000000..26b2f9ae350d1
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/ProducingBean.java
@@ -0,0 +1,25 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar;
+
+import java.time.Duration;
+import java.util.concurrent.Flow;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import org.eclipse.microprofile.reactive.messaging.Outgoing;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+
+@ApplicationScoped
+public class ProducingBean {
+
+ @Outgoing("source")
+ public Flow.Publisher generate() {
+ return Multi.createFrom().range(1, 11)
+ .map(Integer::longValue)
+ .map(i -> i * 2)
+ .onItem()
+ .transformToUniAndConcatenate(l -> Uni.createFrom().item(l).onItem().delayIt().by(Duration.ofMillis(10)));
+ }
+
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/PulsarAuthTest.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/PulsarAuthTest.java
new file mode 100644
index 0000000000000..d4205ddbc3d30
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/PulsarAuthTest.java
@@ -0,0 +1,28 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar;
+
+import static org.awaitility.Awaitility.await;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.restassured.RestAssured;
+
+public class PulsarAuthTest {
+
+ @RegisterExtension
+ static final QuarkusUnitTest config = new QuarkusUnitTest().setArchiveProducer(
+ () -> ShrinkWrap.create(JavaArchive.class)
+ .addClasses(ConsumingBean.class, ProducingBean.class, TestResource.class))
+ .withConfigurationResource("application-secured.properties");
+
+ @Test
+ public void test() {
+ await().until(() -> {
+ String value = RestAssured.get("/last").asString();
+ return value.equalsIgnoreCase("20");
+ });
+ }
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/TestResource.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/TestResource.java
new file mode 100644
index 0000000000000..492ee53a8fdb7
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/TestResource.java
@@ -0,0 +1,18 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar;
+
+import jakarta.inject.Inject;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+
+@Path("/last")
+public class TestResource {
+
+ @Inject
+ ConsumingBean bean;
+
+ @GET
+ public long getLast() {
+ return bean.get();
+ }
+
+}
diff --git a/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/DefaultSchemaConfigTest.java b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/DefaultSchemaConfigTest.java
new file mode 100644
index 0000000000000..76ae6bfd31286
--- /dev/null
+++ b/extensions/smallrye-reactive-messaging-pulsar/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/pulsar/deployment/DefaultSchemaConfigTest.java
@@ -0,0 +1,2042 @@
+package io.quarkus.smallrye.reactivemessaging.pulsar.deployment;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.groups.Tuple.tuple;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Supplier;
+
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Inject;
+
+import org.apache.avro.specific.AvroGenerated;
+import org.apache.pulsar.client.api.Messages;
+import org.apache.pulsar.client.api.Schema;
+import org.assertj.core.groups.Tuple;
+import org.eclipse.microprofile.config.Config;
+import org.eclipse.microprofile.config.spi.ConfigProviderResolver;
+import org.eclipse.microprofile.reactive.messaging.Channel;
+import org.eclipse.microprofile.reactive.messaging.Emitter;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.eclipse.microprofile.reactive.messaging.Message;
+import org.eclipse.microprofile.reactive.messaging.Outgoing;
+import org.eclipse.microprofile.reactive.streams.operators.ProcessorBuilder;
+import org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder;
+import org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder;
+import org.jboss.jandex.IndexView;
+import org.jboss.jandex.Indexer;
+import org.jboss.jandex.Type;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.reactivestreams.Processor;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.builditem.RunTimeConfigurationDefaultBuildItem;
+import io.quarkus.deployment.recording.RecorderContext;
+import io.quarkus.pulsar.SchemaProviderRecorder;
+import io.quarkus.runtime.RuntimeValue;
+import io.quarkus.smallrye.reactivemessaging.deployment.items.ConnectorManagedChannelBuildItem;
+import io.smallrye.common.annotation.Identifier;
+import io.smallrye.config.SmallRyeConfigBuilder;
+import io.smallrye.config.common.MapBackedConfigSource;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import io.smallrye.reactive.messaging.MutinyEmitter;
+import io.smallrye.reactive.messaging.pulsar.OutgoingMessage;
+import io.smallrye.reactive.messaging.pulsar.PulsarBatchMessage;
+import io.smallrye.reactive.messaging.pulsar.PulsarMessage;
+import io.smallrye.reactive.messaging.pulsar.transactions.PulsarTransactions;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.JsonObject;
+
+public class DefaultSchemaConfigTest {
+ private static void doTest(Tuple[] expectations, Class... classesToIndex) {
+ doTest(null, expectations, Map.of(), classesToIndex);
+ }
+
+ private static void doTest(Tuple[] expectations, Map generatedSchemas, Class... classesToIndex) {
+ doTest(null, expectations, generatedSchemas, classesToIndex);
+ }
+
+ private static void doTest(Config customConfig, Tuple[] expectations, Map generatedSchemas,
+ Class... classesToIndex) {
+ List syntheticBeans = new ArrayList<>();
+ List configs = new ArrayList<>();
+
+ List> classes = new ArrayList<>(Arrays.asList(classesToIndex));
+ classes.add(Incoming.class);
+ DefaultSchemaDiscoveryState discovery = new DefaultSchemaDiscoveryState(index(classes)) {
+ @Override
+ Config getConfig() {
+ return customConfig != null ? customConfig : super.getConfig();
+ }
+
+ @Override
+ boolean isPulsarConnector(List list, boolean incoming, String channelName) {
+ return true;
+ }
+ };
+ RecorderContext rcMock = Mockito.mock(RecorderContext.class);
+ Mockito.when(rcMock.classProxy(Mockito.anyString())).thenAnswer(a -> Class.forName(a.getArgument(0)));
+ SyntheticBeanBuilder syntheticBean = new SyntheticBeanBuilder(syntheticBeans::add,
+ new SchemaProviderRecorder(), rcMock) {
+ @Override
+ void produceSyntheticBeanSchema(BuildProducer syntheticBeanBuildItem,
+ RuntimeValue runtimeValue, String schemaId, Type type) {
+ // no-op
+ }
+
+ @Override
+ void produceSyntheticBeanSchema(BuildProducer syntheticBeanBuildItem,
+ Supplier supplier, String schemaId, Type type) {
+ // no-op
+ }
+
+ @Override
+ String generateId(Type type, String targetType) {
+ // remove the random bits
+ return type.name().withoutPackagePrefix() + targetType + "Schema";
+ }
+ };
+
+ try {
+ new PulsarSchemaDiscoveryProcessor().discoverDefaultSerdeConfig(discovery,
+ Collections.emptyList(),
+ configs::add, syntheticBean);
+
+ assertThat(configs)
+ .extracting(RunTimeConfigurationDefaultBuildItem::getKey, RunTimeConfigurationDefaultBuildItem::getValue)
+ .containsExactlyInAnyOrder(expectations);
+
+ assertThat(syntheticBean.alreadyGeneratedSchema).containsExactlyInAnyOrderEntriesOf(generatedSchemas);
+ } finally {
+ // must not leak the Config instance associated to the system classloader
+ if (customConfig == null) {
+ ConfigProviderResolver.instance().releaseConfig(discovery.getConfig());
+ }
+ }
+ }
+
+ private static IndexView index(List> classes) {
+ Indexer indexer = new Indexer();
+ for (Class clazz : classes) {
+ try {
+ try (InputStream stream = DefaultSchemaConfigTest.class.getClassLoader()
+ .getResourceAsStream(clazz.getName().replace('.', '/') + ".class")) {
+ indexer.index(stream);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return indexer.complete();
+ }
+
+ // ---
+
+ @Test
+ public void stringInLongOut() {
+ // @formatter:off
+ Tuple[] expectations = {
+ tuple("mp.messaging.outgoing.channel1.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel2.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel3.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel4.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel5.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel6.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel7.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel8.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel9.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel10.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel11.schema", "INT64"),
+ tuple("mp.messaging.outgoing.channel12.schema", "INT64"),
+
+ tuple("mp.messaging.incoming.channel13.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel14.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel15.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel16.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel17.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel18.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel19.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel20.schema", "STRING"),
+ tuple("mp.messaging.incoming.channel21.schema", "STRING"),
+
+ tuple("mp.messaging.incoming.channel22.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel23.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel24.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel25.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel26.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel27.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel28.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel29.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel30.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel31.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel32.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel33.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel34.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel35.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel36.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel37.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel38.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel39.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel40.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel41.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel42.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel43.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel44.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel45.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel46.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel47.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel48.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel49.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel50.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel51.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel52.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel53.schema", "INT64"),
+
+ tuple("mp.messaging.incoming.channel54.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel55.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel56.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel57.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel58.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel59.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel60.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel61.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel62.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel63.schema", "INT64"),
+ tuple("mp.messaging.incoming.channel64.schema", "STRING"),
+ tuple("mp.messaging.outgoing.channel65.schema", "INT64"),
+ };
+ // @formatter:on
+
+ doTest(expectations, StringInLongOut.class);
+ }
+
+ private static class StringInLongOut {
+ // @Outgoing
+
+ @Outgoing("channel1")
+ Publisher> method1() {
+ return null;
+ }
+
+ @Outgoing("channel2")
+ Publisher method2() {
+ return null;
+ }
+
+ @Outgoing("channel3")
+ PublisherBuilder> method3() {
+ return null;
+ }
+
+ @Outgoing("channel4")
+ PublisherBuilder method4() {
+ return null;
+ }
+
+ @Outgoing("channel5")
+ Multi> method5() {
+ return null;
+ }
+
+ @Outgoing("channel6")
+ Multi method6() {
+ return null;
+ }
+
+ @Outgoing("channel7")
+ Message method7() {
+ return null;
+ }
+
+ @Outgoing("channel8")
+ Long method8() {
+ return null;
+ }
+
+ @Outgoing("channel9")
+ CompletionStage> method9() {
+ return null;
+ }
+
+ @Outgoing("channel10")
+ CompletionStage method10() {
+ return null;
+ }
+
+ @Outgoing("channel11")
+ Uni> method11() {
+ return null;
+ }
+
+ @Outgoing("channel12")
+ Uni method12() {
+ return null;
+ }
+
+ // @Incoming
+
+ @Incoming("channel13")
+ Subscriber> method13() {
+ return null;
+ }
+
+ @Incoming("channel14")
+ Subscriber method14() {
+ return null;
+ }
+
+ @Incoming("channel15")
+ SubscriberBuilder, Void> method15() {
+ return null;
+ }
+
+ @Incoming("channel16")
+ SubscriberBuilder method16() {
+ return null;
+ }
+
+ @Incoming("channel17")
+ void method17(String msg) {
+ }
+
+ @Incoming("channel18")
+ CompletionStage method18(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel19")
+ CompletionStage method19(String payload) {
+ return null;
+ }
+
+ @Incoming("channel20")
+ Uni method20(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel21")
+ Uni method21(String payload) {
+ return null;
+ }
+
+ // @Incoming @Outgoing
+
+ @Incoming("channel22")
+ @Outgoing("channel23")
+ Processor, Message> method22() {
+ return null;
+ }
+
+ @Incoming("channel24")
+ @Outgoing("channel25")
+ Processor method23() {
+ return null;
+ }
+
+ @Incoming("channel26")
+ @Outgoing("channel27")
+ ProcessorBuilder, Message> method24() {
+ return null;
+ }
+
+ @Incoming("channel28")
+ @Outgoing("channel29")
+ ProcessorBuilder method25() {
+ return null;
+ }
+
+ @Incoming("channel30")
+ @Outgoing("channel31")
+ Publisher> method26(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel32")
+ @Outgoing("channel33")
+ Publisher method27(String payload) {
+ return null;
+ }
+
+ @Incoming("channel34")
+ @Outgoing("channel35")
+ PublisherBuilder> method28(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel36")
+ @Outgoing("channel37")
+ PublisherBuilder method29(String payload) {
+ return null;
+ }
+
+ @Incoming("channel38")
+ @Outgoing("channel39")
+ Multi> method30(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel40")
+ @Outgoing("channel41")
+ Multi method31(String payload) {
+ return null;
+ }
+
+ @Incoming("channel42")
+ @Outgoing("channel43")
+ Message method32(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel44")
+ @Outgoing("channel45")
+ Long method33(String payload) {
+ return null;
+ }
+
+ @Incoming("channel46")
+ @Outgoing("channel47")
+ CompletionStage> method34(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel48")
+ @Outgoing("channel49")
+ CompletionStage method35(String payload) {
+ return null;
+ }
+
+ @Incoming("channel50")
+ @Outgoing("channel51")
+ Uni> method36(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel52")
+ @Outgoing("channel53")
+ Uni method37(String payload) {
+ return null;
+ }
+
+ // @Incoming @Outgoing stream manipulation
+
+ @Incoming("channel54")
+ @Outgoing("channel55")
+ Publisher> method38(Publisher> msg) {
+ return null;
+ }
+
+ @Incoming("channel56")
+ @Outgoing("channel57")
+ Publisher method39(Publisher payload) {
+ return null;
+ }
+
+ @Incoming("channel58")
+ @Outgoing("channel59")
+ PublisherBuilder> method40(PublisherBuilder> msg) {
+ return null;
+ }
+
+ @Incoming("channel60")
+ @Outgoing("channel61")
+ PublisherBuilder method41(PublisherBuilder payload) {
+ return null;
+ }
+
+ @Incoming("channel62")
+ @Outgoing("channel63")
+ Multi> method42(Multi> msg) {
+ return null;
+ }
+
+ @Incoming("channel64")
+ @Outgoing("channel65")
+ Multi method43(Multi payload) {
+ return null;
+ }
+ }
+
+ // ---
+
+ @Test
+ public void byteArrayInAvroDtoOut() {
+ // @formatter:off
+ Tuple[] expectations = {
+ tuple("mp.messaging.outgoing.channel1.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel2.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel3.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel4.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel5.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel6.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel7.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel8.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel9.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel10.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel11.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.outgoing.channel12.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+
+ tuple("mp.messaging.incoming.channel13.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel14.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel15.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel16.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel17.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel18.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel19.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel20.schema", "BYTES"),
+ tuple("mp.messaging.incoming.channel21.schema", "BYTES"),
+
+ tuple("mp.messaging.incoming.channel22.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel23.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel24.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel25.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel26.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel27.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel28.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel29.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel30.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel31.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel32.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel33.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel34.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel35.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel36.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel37.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel38.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel39.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel40.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel41.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel42.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel43.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel44.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel45.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel46.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel47.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel48.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel49.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel50.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel51.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel52.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel53.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+
+ tuple("mp.messaging.incoming.channel54.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel55.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel56.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel57.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel58.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel59.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel60.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel61.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel62.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel63.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ tuple("mp.messaging.incoming.channel64.schema", "BYTES"),
+ tuple("mp.messaging.outgoing.channel65.schema", "DefaultSchemaConfigTest$AvroDtoAVROSchema"),
+ };
+ Map generatedSchemas = Map.of(
+ "io.quarkus.smallrye.reactivemessaging.pulsar.deployment.DefaultSchemaConfigTest$AvroDto",
+ "DefaultSchemaConfigTest$AvroDtoAVROSchema"
+ );
+ // @formatter:on
+
+ doTest(expectations, generatedSchemas, AvroDto.class, ByteArrayInAvroDtoOut.class);
+ }
+
+ // simulating an Avro-generated class, the autodetection code only looks for this annotation
+ @AvroGenerated
+ private static class AvroDto {
+ }
+
+ private static class ByteArrayInAvroDtoOut {
+ // @Outgoing
+
+ @Outgoing("channel1")
+ Publisher> method1() {
+ return null;
+ }
+
+ @Outgoing("channel2")
+ Publisher method2() {
+ return null;
+ }
+
+ @Outgoing("channel3")
+ PublisherBuilder> method3() {
+ return null;
+ }
+
+ @Outgoing("channel4")
+ PublisherBuilder method4() {
+ return null;
+ }
+
+ @Outgoing("channel5")
+ Multi> method5() {
+ return null;
+ }
+
+ @Outgoing("channel6")
+ Multi method6() {
+ return null;
+ }
+
+ @Outgoing("channel7")
+ Message method7() {
+ return null;
+ }
+
+ @Outgoing("channel8")
+ AvroDto method8() {
+ return null;
+ }
+
+ @Outgoing("channel9")
+ CompletionStage> method9() {
+ return null;
+ }
+
+ @Outgoing("channel10")
+ CompletionStage method10() {
+ return null;
+ }
+
+ @Outgoing("channel11")
+ Uni> method11() {
+ return null;
+ }
+
+ @Outgoing("channel12")
+ Uni method12() {
+ return null;
+ }
+
+ // @Incoming
+
+ @Incoming("channel13")
+ Subscriber> method13() {
+ return null;
+ }
+
+ @Incoming("channel14")
+ Subscriber method14() {
+ return null;
+ }
+
+ @Incoming("channel15")
+ SubscriberBuilder, Void> method15() {
+ return null;
+ }
+
+ @Incoming("channel16")
+ SubscriberBuilder method16() {
+ return null;
+ }
+
+ @Incoming("channel17")
+ void method17(byte[] msg) {
+ }
+
+ @Incoming("channel18")
+ CompletionStage method18(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel19")
+ CompletionStage method19(byte[] payload) {
+ return null;
+ }
+
+ @Incoming("channel20")
+ Uni method20(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel21")
+ Uni method21(byte[] payload) {
+ return null;
+ }
+
+ // @Incoming @Outgoing
+
+ @Incoming("channel22")
+ @Outgoing("channel23")
+ Processor, Message> method22() {
+ return null;
+ }
+
+ @Incoming("channel24")
+ @Outgoing("channel25")
+ Processor method23() {
+ return null;
+ }
+
+ @Incoming("channel26")
+ @Outgoing("channel27")
+ ProcessorBuilder, Message> method24() {
+ return null;
+ }
+
+ @Incoming("channel28")
+ @Outgoing("channel29")
+ ProcessorBuilder method25() {
+ return null;
+ }
+
+ @Incoming("channel30")
+ @Outgoing("channel31")
+ Publisher> method26(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel32")
+ @Outgoing("channel33")
+ Publisher method27(byte[] payload) {
+ return null;
+ }
+
+ @Incoming("channel34")
+ @Outgoing("channel35")
+ PublisherBuilder> method28(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel36")
+ @Outgoing("channel37")
+ PublisherBuilder method29(byte[] payload) {
+ return null;
+ }
+
+ @Incoming("channel38")
+ @Outgoing("channel39")
+ Multi> method30(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel40")
+ @Outgoing("channel41")
+ Multi method31(byte[] payload) {
+ return null;
+ }
+
+ @Incoming("channel42")
+ @Outgoing("channel43")
+ Message method32(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel44")
+ @Outgoing("channel45")
+ AvroDto method33(byte[] payload) {
+ return null;
+ }
+
+ @Incoming("channel46")
+ @Outgoing("channel47")
+ CompletionStage> method34(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel48")
+ @Outgoing("channel49")
+ CompletionStage method35(byte[] payload) {
+ return null;
+ }
+
+ @Incoming("channel50")
+ @Outgoing("channel51")
+ Uni> method36(Message msg) {
+ return null;
+ }
+
+ @Incoming("channel52")
+ @Outgoing("channel53")
+ Uni method37(byte[] payload) {
+ return null;
+ }
+
+ // @Incoming @Outgoing stream manipulation
+
+ @Incoming("channel54")
+ @Outgoing("channel55")
+ Publisher> method38(Publisher> msg) {
+ return null;
+ }
+
+ @Incoming("channel56")
+ @Outgoing("channel57")
+ Publisher method39(Publisher payload) {
+ return null;
+ }
+
+ @Incoming("channel58")
+ @Outgoing("channel59")
+ PublisherBuilder> method40(PublisherBuilder> msg) {
+ return null;
+ }
+
+ @Incoming("channel60")
+ @Outgoing("channel61")
+ PublisherBuilder method41(PublisherBuilder payload) {
+ return null;
+ }
+
+ @Incoming("channel62")
+ @Outgoing("channel63")
+ Multi> method42(Multi> msg) {
+ return null;
+ }
+
+ @Incoming("channel64")
+ @Outgoing("channel65")
+ Multi method43(Multi payload) {
+ return null;
+ }
+ }
+
+ // ---
+
+ @Test
+ public void jacksonDtoInVertxJsonObjectOut() {
+ // @formatter:off
+ Tuple[] expectations = {
+ tuple("mp.messaging.outgoing.channel1.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel2.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel3.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel4.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel5.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel6.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel7.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel8.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel9.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel10.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel11.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.outgoing.channel12.schema", "JsonObjectJSON_OBJECTSchema"),
+
+ tuple("mp.messaging.incoming.channel13.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel14.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel15.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel16.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel17.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel18.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel19.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel20.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.incoming.channel21.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+
+ tuple("mp.messaging.incoming.channel22.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel23.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel24.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel25.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel26.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel27.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel28.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel29.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel30.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel31.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel32.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel33.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel34.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel35.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel36.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel37.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel38.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel39.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel40.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel41.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel42.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel43.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel44.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel45.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel46.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel47.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel48.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel49.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel50.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel51.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel52.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel53.schema", "JsonObjectJSON_OBJECTSchema"),
+
+ tuple("mp.messaging.incoming.channel54.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel55.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel56.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel57.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel58.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel59.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel60.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel61.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel62.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel63.schema", "JsonObjectJSON_OBJECTSchema"),
+ tuple("mp.messaging.incoming.channel64.schema", "DefaultSchemaConfigTest$JacksonDtoJSONSchema"),
+ tuple("mp.messaging.outgoing.channel65.schema", "JsonObjectJSON_OBJECTSchema"),
+ };
+ var generatedSchemas = Map.of(
+ "io.quarkus.smallrye.reactivemessaging.pulsar.deployment.DefaultSchemaConfigTest$JacksonDto",
+ "DefaultSchemaConfigTest$JacksonDtoJSONSchema",
+ "io.vertx.core.json.JsonObject", "JsonObjectJSON_OBJECTSchema"
+ );
+ // @formatter:on
+
+ doTest(expectations, generatedSchemas, JacksonDto.class, JacksonDtoInVertxJsonObjectOut.class);
+
+ }
+
+ static class JacksonDto {
+ }
+
+ private static class JacksonDtoInVertxJsonObjectOut {
+ // @Outgoing
+
+ @Outgoing("channel1")
+ Publisher> method1() {
+ return null;
+ }
+
+ @Outgoing("channel2")
+ Publisher method2() {
+ return null;
+ }
+
+ @Outgoing("channel3")
+ PublisherBuilder> method3() {
+ return null;
+ }
+
+ @Outgoing("channel4")
+ PublisherBuilder