Package org.eclipse.microprofile.reactive.messaging.spi

Interface IncomingConnectorFactory

  • All Superinterfaces:
    ConnectorFactory
    All Known Implementing Classes:
    AmqpConnector, CamelConnector, InMemoryConnector, JmsConnector, KafkaConnector, MqttConnector, PubSubConnector, RabbitMQConnector
    public interface IncomingConnectorFactory
extends ConnectorFactory
    SPI used to implement a connector managing a source of messages for a specific transport. For example, to handle the consumption of records from Kafka, the reactive messaging extension would need to implement a bean implementing this interface. This bean is called for every channel that needs to be created for this specific transport (so Kafka in this case). These channels are connected to methods annotated with Incoming.

    The factory is called to create a channel for each configured transport. The configuration is done using MicroProfile Config. The following snippet gives an example for a hypothetical Kafka connector: 

     mp.messaging.incoming.my-channel.topic=my-topic
 mp.messaging.connector.acme.kafka.bootstrap.servers=localhost:9092
 ...

    The configuration keys are structured as follows: mp.messaging.[incoming|outgoing].channel-name.attribute or mp.messaging.[connector].connector-name.attribute. Channel names are not expected to contain . so the first occurrence of a . in the channel-name portion of a property terminates the channel name and precedes the attribute name. For connector attributes, the longest string, inclusive of .s, that matches a loadable connector is used as a connector-name. The remainder, after a . separator, is the attribute name. Configuration keys that begin mp.messaging.outgoing} are not used for IncomingConnectorFactory configuration.

    The portion of the key that precedes the attribute acts as a property prefix that has a common structure across all MicroProfile Reactive Messaging configuration properties.

    The channel-name segment in the configuration key corresponds to the name of the channel used in the Incoming annotation: 

     @Incoming("my-channel")
 public void consume(String s) {
     // ...
 }

    The set of attributes depend on the connector and transport layer (for example, bootstrap.servers is Kafka specific). The connector attribute indicates the name of the connector. It will be matched to the value returned by the Connector qualifier used on the relevant IncomingConnectorFactory bean implementation. This is how a reactive messaging implementation looks for the specific IncomingConnectorFactory required for a channel. Any mp.messaging.connector attributes for the channel's connector are also included in the set of relevant attributes. Where an attribute is present for both a channel and its connector the value of the channel specific attribute will take precedence. In the previous configuration, the reactive messaging implementation would need to find the IncomingConnectorFactory qualified using the Connector qualifier with the value acme.kafka class to create the my-channel channel. Note that if the connector cannot be found, the deployment must be failed with a DeploymentException.

    The getPublisherBuilder(Config) is called for every channel that needs to be created. The Config object passed to the method contains a subset of the global configuration, and with the prefixes removed. So for the previous configuration, it would be: 

     bootstrap.servers = localhost:9092
 topic = my-topic

    In this example, if topic was missing as a configuration property, the Kafka connector would be at liberty to default to the channel name indicated in the annotation as the Kafka topic. Such connector specific behaviours are outside the scope of this specification.

    So the connector implementation can retrieve the value with Config.getValue(String, Class) and Config.getOptionalValue(String, Class).

    If the configuration is invalid, the getPublisherBuilder(Config) method must throw an IllegalArgumentException, caught by the reactive messaging implementation and failing the deployment by throwing a DeploymentException wrapping the exception.

    Note that a Reactive Messaging implementation must support the configuration format described here. Implementations are free to provide additional support for other approaches.

    • Method Detail

      • getPublisherBuilder

        org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder<? extends Message<?>> getPublisherBuilder​(org.eclipse.microprofile.config.Config config)
        Creates a channel for the given configuration. The channel's configuration is associated with a specific connector, using the Connector qualifier's parameter indicating a key to which IncomingConnectorFactory to use.

        Note that the connection to the transport or broker is generally postponed until the subscription occurs.

        Parameters:
        config - the configuration, must not be null, must contain the ConnectorFactory.CHANNEL_NAME_ATTRIBUTE attribute.
        Returns:
        the created PublisherBuilder, will not be null.
        Throws:
        IllegalArgumentException - if the configuration is invalid.
        NoSuchElementException - if the configuration does not contain an expected attribute.