Package com.hazelcast.jet.pipeline

Class JmsSourceBuilder

java.lang.Object
com.hazelcast.jet.pipeline.JmsSourceBuilder
public final class JmsSourceBuilder extends Object
See Sources.jmsQueueBuilder(com.hazelcast.function.SupplierEx<? extends jakarta.jms.ConnectionFactory>) or Sources.jmsTopicBuilder(com.hazelcast.function.SupplierEx<? extends jakarta.jms.ConnectionFactory>).
Since:
Jet 3.0

  • Method Details

    • connectionParams

      @Nonnull public JmsSourceBuilder connectionParams(@Nullable String username, @Nullable String password)
      Sets the connection parameters. If connectionFn(FunctionEx) is set, these parameters are ignored.
      Returns:
      this instance for fluent API

    • connectionFn

      @Nonnull public JmsSourceBuilder connectionFn(@Nullable FunctionEx<? super jakarta.jms.ConnectionFactory,? extends jakarta.jms.Connection> connectionFn)
      Sets the function which creates the connection using the connection factory.

      If not provided, this function is used: 

           connectionFn = factory -> username != null || password != null
         ? factory.createConnection(usernameLocal, passwordLocal)
         : factory.createConnection()
      The user name and password set with connectionParams(java.lang.String, java.lang.String) are used.

      The given function must be stateless.

      Returns:
      this instance for fluent API

    • destinationName

      @Nonnull public JmsSourceBuilder destinationName(@Nullable String destinationName)
      Sets the name of the destination (name of the topic or queue). If consumerFn(FunctionEx) is provided, this parameter is ignored.
      Returns:
      this instance for fluent API

    • consumerFn

      @Nonnull public JmsSourceBuilder consumerFn(@Nullable FunctionEx<? super jakarta.jms.Session,? extends jakarta.jms.MessageConsumer> consumerFn)
      Sets the function which creates the message consumer from session.

      If not provided, Session#createConsumer(destinationName) is used to create the consumer. See destinationName(String).

      If you're consuming a topic and you create a shared consumer, make sure to also call sharedConsumer(true).

      The given function must be stateless.

      Returns:
      this instance for fluent API

    • messageIdFn

      @Nonnull public JmsSourceBuilder messageIdFn(@Nonnull FunctionEx<? super jakarta.jms.Message,?> messageIdFn)
      Configures the function to extract IDs from the messages, if exactly-once guarantee is used. If a lower guarantee is used, this function is not used.

      Make sure the function returns non-null for every message, or the job will fail. The returned object should also implement equals() and hashCode() methods. If you don't have a unique message ID, reduce the guarantee to at-least-once.

      The default is to use Message.getJMSMessageID().

      The given function must be stateless.

      Returns:
      this instance for fluent API

    • maxGuarantee

      @Nonnull public JmsSourceBuilder maxGuarantee(@Nonnull ProcessingGuarantee guarantee)
      Sets the maximum processing guarantee for the source. You can use it to reduce the guarantee of this source compared to the job's guarantee. If you configure a stronger guarantee than the job has, the job's guarantee will be used. Use it if you want to avoid the overhead of acknowledging the messages or storing IDs of seen messages, if you can tolerate duplicated or missed messages.

      If the processing guarantee is NONE, the processor will consume the messages in Session.DUPS_OK_ACKNOWLEDGE mode. If the processing guarantee is other than NONE, the processor will acknowledge messages in transactions in the 2nd phase of the snapshot, that is after all downstream stages fully processed the messages. Additionally, if the processing guarantee is EXACTLY_ONCE, the processor will store message IDs of the unacknowledged messages to the snapshot and should the job fail after the snapshot was successful, but before Jet managed to acknowledge the messages. The stored IDs will be used to filter out the re-delivered messages.

      If you use a non-durable consumer with a topic, the guarantee will not work since the broker doesn't store the messages at all. You can also set the max-guarantee to NONE in this case - the acknowledge operation is ignored anyway. If you didn't specify your own consumerFn(FunctionEx), a non-durable consumer is created for a topic by default.

      The default is ProcessingGuarantee.EXACTLY_ONCE, which means that the source's guarantee will match the job's guarantee.

      Returns:
      this instance for fluent API

    • sharedConsumer

      @Nonnull public JmsSourceBuilder sharedConsumer(boolean isSharedConsumer)
      Specifies whether the MessageConsumer of the JMS topic is shared, that is whether createSharedConsumer() or createSharedDurableConsumer() was used to create it in the consumerFn(FunctionEx).

      If the consumer is not shared, only a single processor on a single member will connect to the broker to receive the messages. If you set this parameter to true for a non-shared consumer, all messages will be emitted on every member, leading to duplicate processing.

      A consumer for a queue is always assumed to be shared, regardless of this setting.

      The default value is false.

      Returns:
      this instance for fluent API

    • build

      @Nonnull public <T> StreamSource<T> build(@Nonnull FunctionEx<? super jakarta.jms.Message,? extends T> projectionFn)
      Creates and returns the JMS StreamSource with the supplied components and the projection function projectionFn.

      The given function must be stateless.

      Type Parameters:
      T - the type of the items the source emits
      Parameters:
      projectionFn - the function which creates output object from each message

    • build

      @Nonnull public StreamSource<jakarta.jms.Message> build()
      Convenience for build(FunctionEx).