Producer

copysourcejavax.jms.ConnectionFactory connectionFactory = server.createConnectionFactory();

copysourceSink<JmsTextMessage, CompletionStage<Done>> jmsSink =
    JmsProducer.sink(
        JmsProducerSettings.create(producerConfig, connectionFactory).withQueue("test"));

CompletionStage<Done> finished =
    Source.from(Arrays.asList("Message A", "Message B"))
        .map(JmsTextMessage::create)
        .runWith(jmsSink, system);

copysourceJmsTextMessage message =
    org.apache.pekko.stream.connectors.jms.JmsTextMessage.create(n.toString())
        .withProperty("Number", n)
        .withProperty("IsOdd", n % 2 == 1)
        .withProperty("IsEven", n % 2 == 0);

copysourceList<JmsTextMessage> msgsIn =
    createTestMessageList().stream()
        .map(
            jmsTextMessage ->
                jmsTextMessage
                    .withHeader(JmsType.create("type"))
                    .withHeader(JmsCorrelationId.create("correlationId"))
                    .withHeader(JmsReplyTo.queue("test-reply"))
                    .withHeader(JmsTimeToLive.create(999, TimeUnit.SECONDS))
                    .withHeader(JmsPriority.create(2))
                    .withHeader(JmsDeliveryMode.create(DeliveryMode.NON_PERSISTENT)))
        .collect(Collectors.toList());

copysourcejavax.jms.ConnectionFactory connectionFactory = server.createConnectionFactory();

Sink<String, CompletionStage<Done>> jmsSink =
    JmsProducer.textSink(
        JmsProducerSettings.create(system, connectionFactory).withQueue("test"));

List<String> in = Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k");
CompletionStage<Done> finished = Source.from(in).runWith(jmsSink, system);

copysourceConnectionFactory connectionFactory = server.createConnectionFactory();

Sink<byte[], CompletionStage<Done>> jmsSink =
    JmsProducer.bytesSink(
        JmsProducerSettings.create(producerConfig, connectionFactory).withQueue("test"));

byte[] in = "ThisIsATest".getBytes(StandardCharsets.UTF_8);
CompletionStage<Done> finished = Source.single(in).runWith(jmsSink, system);

copysourceConnectionFactory connectionFactory = server.createConnectionFactory();

Sink<Map<String, Object>, CompletionStage<Done>> jmsSink =
    JmsProducer.mapSink(
        JmsProducerSettings.create(system, connectionFactory).withQueue("test"));

Map<String, Object> in = new HashMap<>();
in.put("string value", "value");
in.put("int value", 42);
in.put("double value", 43.0);
in.put("short value", (short) 7);
in.put("boolean value", true);
in.put("long value", 7L);
in.put("bytearray", "AStringAsByteArray".getBytes(StandardCharsets.UTF_8));
in.put("byte", (byte) 1);

CompletionStage<Done> finished = Source.single(in).runWith(jmsSink, system);

copysourceActiveMQConnectionFactory connectionFactory =
    (ActiveMQConnectionFactory) server.createConnectionFactory();
connectionFactory.setTrustedPackages(
    Arrays.asList(DummyJavaTests.class.getPackage().getName()));

Sink<java.io.Serializable, CompletionStage<Done>> jmsSink =
    JmsProducer.objectSink(
        JmsProducerSettings.create(system, connectionFactory).withQueue("test"));

java.io.Serializable in = new DummyJavaTests("javaTest");
CompletionStage<Done> finished = Source.single(in).runWith(jmsSink, system);

copysourceConnectionFactory connectionFactory = server.createConnectionFactory();

Flow<JmsTextMessage, JmsTextMessage, JmsProducerStatus> flow =
    JmsProducer.flow(
        JmsProducerSettings.create(system, connectionFactory).withQueue("test"));

List<JmsTextMessage> input = createTestMessageList();

CompletionStage<List<JmsTextMessage>> result =
    Source.from(input).via(flow).runWith(Sink.seq(), system);

copysourceFlow<JmsTextMessage, JmsTextMessage, JmsProducerStatus> flowSink =
    JmsProducer.flow(
        JmsProducerSettings.create(system, connectionFactory).withQueue("test"));

List<JmsTextMessage> input = new ArrayList<>();
for (Integer n : Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)) {
  String queueName = (n % 2 == 0) ? "even" : "odd";
  input.add(JmsTextMessage.create(n.toString()).toQueue(queueName));
}

Source.from(input).via(flowSink).runWith(Sink.seq(), system);

copysourceFlow<JmsEnvelope<String>, JmsEnvelope<String>, JmsProducerStatus> jmsProducer =
    JmsProducer.flexiFlow(
        JmsProducerSettings.create(producerConfig, connectionFactory).withQueue("test"));

List<String> data = Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k");
List<JmsEnvelope<String>> input = new ArrayList<>();
for (String s : data) {
  String passThrough = s;
  input.add(JmsTextMessage.create(s, passThrough));
}

CompletionStage<List<String>> result =
    Source.from(input)
        .via(jmsProducer)
        .map(JmsEnvelope::passThrough)
        .runWith(Sink.seq(), system);

copysourceFlow<JmsEnvelope<String>, JmsEnvelope<String>, JmsProducerStatus> jmsProducer =
    JmsProducer.flexiFlow(
        JmsProducerSettings.create(producerConfig, connectionFactory).withQueue("test"));

List<String> data = Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k");
List<JmsEnvelope<String>> input = new ArrayList<>();
for (String s : data) {
  String passThrough = s;
  input.add(JmsPassThrough.create(passThrough));
}

CompletionStage<List<String>> result =
    Source.from(input)
        .via(jmsProducer)
        .map(JmsEnvelope::passThrough)
        .runWith(Sink.seq(), system);

copysourceConfig producerConfig = config.getConfig(JmsProducerSettings.configPath());
JmsProducerSettings settings =
    JmsProducerSettings.create(producerConfig, new ActiveMQConnectionFactory("broker-url"))
        .withTopic("target-topic")
        .withCredentials(Credentials.create("username", "password"))
        .withConnectionRetrySettings(retrySettings)
        .withSendRetrySettings(sendRetrySettings)
        .withSessionCount(10)
        .withTimeToLive(Duration.ofHours(1));

copysource# Jms Producer Settings
# sets default values
producer {
  # Configure connection retrying by providing settings for ConnectionRetrySettings.
  connection-retry = ${pekko.connectors.jms.connection-retry}
  # Configure re-sending by providing settings for SendRetrySettings.
  send-retry = ${pekko.connectors.jms.send-retry}
  # Credentials to connect to the JMS broker.
  # credentials {
  #   username = "some text"
  #   password = "some text"
  # }
  # "off" to not use any credentials.
  credentials = off
  # Number of parallel sessions to use for sending JMS messages.
  # Increasing the number of parallel sessions increases throughput at the cost of message ordering.
  # While the messages may arrive out of order on the JMS broker, the producer flow outputs messages
  # in the order they are received.
  session-count = 1
  # Time messages should be kept on the JMS broker.
  # This setting can be overridden on individual messages.
  # "off" to not let messages expire.
  time-to-live = off
  # How long the stage should preserve connection status events for the first subscriber before discarding them
  connection-status-subscription-timeout = 5 seconds
}

copysource# Connection Retry Settings
# these set the defaults for Consumer, Producer, and Browse settings
connection-retry {
  # Time allowed to establish and start a connection.
  connect-timeout = 10 seconds
  # Wait time before retrying the connection the first time.
  initial-retry = 100 millis
  # Back-off factor for subsequent retries.
  backoff-factor = 2
  # Back-off factor for subsequent retries.
  max-backoff = 1 minute
  # Maximum number of retries allowed.
  # "infinite", or positive integer
  max-retries = 10
}

copysourceConfig connectionRetryConfig = config.getConfig("pekko.connectors.jms.connection-retry");
// reiterating the values from reference.conf
ConnectionRetrySettings retrySettings =
    ConnectionRetrySettings.create(connectionRetryConfig)
        .withConnectTimeout(Duration.ofSeconds(10))
        .withInitialRetry(Duration.ofMillis(100))
        .withBackoffFactor(2.0)
        .withMaxBackoff(Duration.ofMinutes(1))
        .withMaxRetries(10);

copysource# Send Retry Settings
# these set the defaults for Producer settings
send-retry {
  # Wait time before retrying the first time.
  initial-retry = 20 millis
  # Back-off factor for subsequent retries.
  backoff-factor = 1.5
  # Maximum back-off time allowed, after which all retries will happen after this delay.
  max-backoff = 500 millis
  # Maximum number of retries allowed.
  # "infinite", or positive integer
  max-retries = 10
}

copysourceimport com.typesafe.config.Config;
import scala.Option;

Config sendRetryConfig = config.getConfig("pekko.connectors.jms.send-retry");
// reiterating the values from reference.conf
SendRetrySettings sendRetrySettings =
    SendRetrySettings.create(sendRetryConfig)
        .withInitialRetry(Duration.ofMillis(20))
        .withBackoffFactor(1.5d)
        .withMaxBackoff(Duration.ofMillis(500))
        .withMaxRetries(10);