org.eclipse.microprofile.reactive.messaging.spi

## Interface IncomingConnectorFactory

All Superinterfaces:
  ConnectorFactory

    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.

### Field Summary

### Fields inherited from interface org.eclipse.microprofile.reactive.messaging.spi.ConnectorFactory

`CHANNEL_NAME_ATTRIBUTE, CONNECTOR_ATTRIBUTE, CONNECTOR_PREFIX, INCOMING_PREFIX, OUTGOING_PREFIX`

### Method Summary

All Methods Instance Methods Abstract Methods
`PublisherBuilder<? extends Message<?>>` `getPublisherBuilder(Config config)` Creates a channel for the given configuration.

### Method Detail

#### getPublisherBuilder

    PublisherBuilder<? extends Message<?>> getPublisherBuilder(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.