public static <EventT> Write<EventT> write() {
    return new AutoValue_JmsIO_Write.Builder<EventT>().build();
  }
@Test
  public void testWriteDynamicMessage() throws Exception {
    Connection connection = connectionFactory.createConnection(USERNAME, PASSWORD);
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MessageConsumer consumerOne = session.createConsumer(session.createTopic("Topic_One"));
    MessageConsumer consumerTwo = session.createConsumer(session.createTopic("Topic_Two"));
    ArrayList<TestEvent> data = new ArrayList<>();
    for (int i = 0; i < 50; i++) {
      data.add(new TestEvent("Topic_One", "Message One " + i));
    }
    for (int i = 0; i < 100; i++) {
      data.add(new TestEvent("Topic_Two", "Message Two " + i));
    }
    pipeline
        .apply(Create.of(data))
        .apply(
            JmsIO.<TestEvent>write()
                .withConnectionFactory(connectionFactory)
                .withUsername(USERNAME)
                .withPassword(PASSWORD)
                .withRetryConfiguration(retryConfiguration)
                .withTopicNameMapper(e -> e.getTopicName())
                .withValueMapper(
                    (e, s) -> {
                      try {
                        TextMessage msg = s.createTextMessage();
                        msg.setText(e.getValue());
                        return msg;
                      } catch (JMSException ex) {
                        throw new JmsIOException("Error writing TextMessage", ex);
                      }
                    }));

    pipeline.run();

    int count = 0;
    while (consumerOne.receive(1000) != null) {
      count++;
    }
    assertEquals(50, count);

    count = 0;
    while (consumerTwo.receive(1000) != null) {
      count++;
    }
    assertEquals(100, count);
  }