[BACKLOG-21800] JMS Consumer. Configurable jms destination

Renamed JmsContextProvider to JmsProvider, since it supplies
both context and destination.
Moved type enums into JmsProvider.

https://jira.pentaho.com/browse/BACKLOG-21800

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/JmsConstants.java

@@ -0,0 +1,6 @@
+package org.pentaho.di.trans.step.jms;
+
+public interface JmsConstants {
+
+  Class PKG = JmsConstants.class;
+}

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/JmsConsumerMeta.java

@@ -29,7 +29,7 @@
 import org.pentaho.di.trans.step.StepDataInterface;
 import org.pentaho.di.trans.step.StepInterface;
 import org.pentaho.di.trans.step.StepMeta;
-import org.pentaho.di.trans.step.jms.context.JmsContextProvider;
+import org.pentaho.di.trans.step.jms.context.JmsProvider;
 
 import java.util.List;
 
@@ -38,8 +38,8 @@
   description = "", categoryDescription = "Streaming" )
 public class JmsConsumerMeta extends JmsMeta {
 
-  public JmsConsumerMeta( List<JmsContextProvider> jmsContextProviders ) {
-    super( jmsContextProviders );
+  public JmsConsumerMeta( List<JmsProvider> jmsProviders ) {
+    super( jmsProviders );
   }
 
   @SuppressWarnings( "deprecated" )

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/JmsMeta.java

@@ -23,8 +23,6 @@
 package org.pentaho.di.trans.step.jms;
 
 
-import com.ibm.mq.jms.MQQueue;
-
 import javax.jms.JMSContext;
 
 import org.pentaho.di.core.ObjectLocationSpecificationMethod;
@@ -37,22 +35,24 @@
 import org.pentaho.di.core.util.serialization.Sensitive;
 import org.pentaho.di.core.variables.VariableSpace;
 import org.pentaho.di.trans.step.StepDataInterface;
-import org.pentaho.di.trans.step.jms.context.JmsContextProvider;
+import org.pentaho.di.trans.step.jms.context.JmsProvider;
 import org.pentaho.di.trans.streaming.common.BaseStreamStepMeta;
 
 import javax.jms.Destination;
-import javax.jms.JMSException;
 import java.util.List;
 import java.util.Objects;
 
+import static org.pentaho.di.trans.step.jms.context.JmsProvider.ConnectionType.WEBSPHERE;
+import static org.pentaho.di.trans.step.jms.context.JmsProvider.DestinationType.QUEUE;
+
 
 @InjectionSupported ( localizationPrefix = "IBMMQConsumerMeta.Injection." )
 public abstract class JmsMeta extends BaseStreamStepMeta {
 
   //TODO move these props to a container pojo that can be added to both
   // the consumer and producer metas, since BaseStreamStepMeta is just consumer.
 
-  @Injection ( name = "DESTINATION" ) public String destinationName = null;
+  @Injection ( name = "DESTINATION" ) public String destinationName = "DEV.QUEUE.1";
 
   @Injection ( name = "URL" ) public String url = "mq://10.177.178.135:1414/QM1?channel=DEV.APP.SVRCONN";
 
@@ -65,26 +65,19 @@ public abstract class JmsMeta extends BaseStreamStepMeta {
 
   @Injection ( name = "USE_JNDI" ) public boolean useJndi = false;
 
-  @Injection ( name = "CONNECTION_TYPE" ) public String connectionType = "WEBSPHERE";
+  @Injection ( name = "CONNECTION_TYPE" ) public String connectionType = WEBSPHERE.name();
+
+  @Injection ( name = "DESTINATION_TYPE" ) public String destinationType = QUEUE.name();
 
 
-  protected final List<JmsContextProvider> jmsContextProviders;
+  protected final List<JmsProvider> jmsProviders;
 
-  protected JmsMeta( List<JmsContextProvider> jmsContextProviders ) {
+  protected JmsMeta( List<JmsProvider> jmsProviders ) {
     super();
-    this.jmsContextProviders = jmsContextProviders;
+    this.jmsProviders = jmsProviders;
     setSpecificationMethod( ObjectLocationSpecificationMethod.FILENAME );
   }
 
-
-  public Destination getDestination() {
-    try {
-      return new MQQueue( "DEV.QUEUE.1" );
-    } catch ( JMSException e ) {
-      throw new RuntimeException( e );
-    }
-  }
-
   @Override public RowMeta getRowMeta( String s, VariableSpace variableSpace ) throws KettleStepException {
     RowMeta rowMeta = new RowMeta();
     rowMeta.addValueMeta( new ValueMetaString( "message" ) );
@@ -95,11 +88,19 @@ public Destination getDestination() {
     return new GenericStepData();
   }
 
+  Destination getDestination() {
+    return getJmsProvider().getDestination( this );
+  }
+
   JMSContext getJmsContext() {
-    return jmsContextProviders.stream()
-      .map( prov -> prov.get( this ) )
+    return getJmsProvider().getContext( this );
+  }
+
+  private JmsProvider getJmsProvider() {
+    return jmsProviders.stream()
+      .filter( prov -> prov.supports( JmsProvider.ConnectionType.valueOf( connectionType ) ) )
       .filter( Objects::nonNull )
       .findFirst()
-      .orElseThrow( () -> new RuntimeException( "FIXME" ) ); // TODO
+      .orElseThrow( () -> new RuntimeException( "FIXME" ) );
   }
 }

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/JmsStreamSource.java

@@ -35,7 +35,6 @@
 
 public class JmsStreamSource extends BlockingQueueStreamSource<List<Object>> {
 
-
   private final JmsMeta meta;
   private JMSConsumer consumer;
 
@@ -45,9 +44,7 @@ protected JmsStreamSource( BaseStreamStep streamStep, JmsMeta meta ) {
   }
 
   @Override public void open() {
-
-    consumer = meta.getJmsContext()
-      .createConsumer( meta.getDestination() );
+    consumer = meta.getJmsContext().createConsumer( meta.getDestination() );
     consumer.setMessageListener( ( message ) -> {
       try {
         acceptRows( singletonList( of( message.getBody( Object.class ) ) ) );

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/context/JmsContextProvider.java → plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/context/JmsProvider.java

@@ -24,8 +24,29 @@
 
 import org.pentaho.di.trans.step.jms.JmsMeta;
 
+import javax.jms.Destination;
 import javax.jms.JMSContext;
 
-public interface JmsContextProvider {
-  JMSContext get( JmsMeta meta );
+
+public interface JmsProvider {
+
+  boolean supports( ConnectionType type );
+
+  JMSContext getContext( JmsMeta meta );
+
+  Destination getDestination( JmsMeta meta );
+
+
+  enum ConnectionType {
+    WEBSPHERE,
+    ACTIVEMQ,
+    JNDI
+  }
+
+  enum DestinationType {
+    QUEUE,
+    TOPIC
+  }
+
 }
+

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/context/WebsphereMQContextProvider.java → plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/context/WebsphereMQProvider.java

@@ -24,47 +24,68 @@
 
 
 import com.ibm.mq.jms.MQConnectionFactory;
+import com.ibm.mq.jms.MQQueue;
+import com.ibm.mq.jms.MQQueueConnectionFactory;
+import com.ibm.mq.jms.MQTopic;
+import com.ibm.mq.jms.MQTopicConnectionFactory;
 import com.ibm.msg.client.wmq.WMQConstants;
 import org.pentaho.di.core.variables.VariableSpace;
 import org.pentaho.di.core.variables.Variables;
-import org.pentaho.di.trans.step.jms.JmsConnectionType;
+import org.pentaho.di.trans.step.jms.JmsConstants;
 import org.pentaho.di.trans.step.jms.JmsMeta;
 
+import javax.jms.Destination;
 import javax.jms.JMSContext;
 import javax.jms.JMSException;
-
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import static com.google.common.base.Preconditions.checkNotNull;
 import static java.util.regex.Pattern.compile;
-import static org.pentaho.di.trans.step.jms.JmsConnectionType.WEBSPHERE;
+import static org.pentaho.di.i18n.BaseMessages.getString;
+import static org.pentaho.di.trans.step.jms.context.JmsProvider.ConnectionType.WEBSPHERE;
+import static org.pentaho.di.trans.step.jms.context.JmsProvider.DestinationType.QUEUE;
 
-public class WebsphereMQContextProvider implements JmsContextProvider {
+public class WebsphereMQProvider implements JmsProvider {
 
+  @Override public boolean supports( ConnectionType type ) {
+    return type == WEBSPHERE;
+  }
 
-  @Override public JMSContext get( JmsMeta meta ) {
-    if ( JmsConnectionType.valueOf( meta.connectionType ).equals( WEBSPHERE ) ) {
+  @Override public JMSContext getContext( JmsMeta meta ) {
 
-      MQUrlResolver resolver = new MQUrlResolver( meta, new Variables() );
+    MQUrlResolver resolver = new MQUrlResolver( meta, new Variables() );
 
-      MQConnectionFactory mqConnectionFactory = new MQConnectionFactory();
-      mqConnectionFactory.setHostName( resolver.host );
-      try {
-        mqConnectionFactory.setPort( resolver.port );
-        mqConnectionFactory.setBrokerQueueManager( resolver.queueManager );
-        mqConnectionFactory.setQueueManager( "QM1" );
-        mqConnectionFactory.setChannel( resolver.channel );
-        mqConnectionFactory.setTransportType( WMQConstants.WMQ_CM_CLIENT );
-      } catch ( JMSException e ) {
-        e.printStackTrace();
-      }
-      return mqConnectionFactory.createContext( meta.username, meta.password );
+    MQConnectionFactory connFactory = isQueue( meta )
+      ? new MQQueueConnectionFactory() : new MQTopicConnectionFactory();
+
+    connFactory.setHostName( resolver.host );
+    try {
+      connFactory.setPort( resolver.port );
+      connFactory.setQueueManager( resolver.queueManager );
+      connFactory.setChannel( resolver.channel );
+      connFactory.setTransportType( WMQConstants.WMQ_CM_CLIENT );
+    } catch ( JMSException e ) {
+      throw new RuntimeException( e );
     }
-    return null;
+    return connFactory.createContext( meta.username, meta.password );
+  }
+
+  @Override public Destination getDestination( JmsMeta meta ) {
+    checkNotNull( meta.destinationName, getString( JmsConstants.PKG, "JmsWebsphereMQ.DestinationNameRequired" ) );
+    try {
+      return isQueue( meta ) ? new MQQueue( meta.destinationName ) : new MQTopic( meta.destinationName );
+    } catch ( JMSException e ) {
+      throw new RuntimeException( e );
+    }
+  }
+
+  private boolean isQueue( JmsMeta meta ) {
+    return DestinationType.valueOf( meta.destinationType ).equals( QUEUE );
   }
 
 
-  class MQUrlResolver {
+  static class MQUrlResolver {
     private final JmsMeta meta;
     private final Pattern pattern;
 

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/ui/ConnectionForm.java

@@ -44,10 +44,10 @@
 
 import static com.google.common.collect.ImmutableList.of;
 import static org.pentaho.di.i18n.BaseMessages.getString;
+import static org.pentaho.di.trans.step.jms.JmsConstants.PKG;
 import static org.pentaho.di.ui.trans.step.BaseStreamingDialog.INPUT_WIDTH;
 
 public class ConnectionForm {
-  private static Class<?> PKG = ConnectionForm.class;
 
   private final Composite parentComponent;
   JmsConsumerMeta jmsMeta;

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/ui/DestinationForm.java

@@ -34,9 +34,9 @@
 import org.pentaho.di.ui.core.widget.TextVar;
 
 import static org.pentaho.di.i18n.BaseMessages.getString;
+import static org.pentaho.di.trans.step.jms.JmsConstants.PKG;
 
 public class DestinationForm {
-  private static Class<?> PKG = DestinationForm.class;
 
   private final Composite parentComponent;
   private final PropsUI props;

plugins/streaming/impls/jms/src/main/java/org/pentaho/di/trans/step/jms/ui/JmsConsumerDialog.java

@@ -33,8 +33,9 @@
 import org.pentaho.di.trans.streaming.common.BaseStreamStepMeta;
 import org.pentaho.di.ui.trans.step.BaseStreamingDialog;
 
+import static org.pentaho.di.trans.step.jms.JmsConstants.PKG;
+
 public class JmsConsumerDialog extends BaseStreamingDialog {
-  private static Class<?> PKG = JmsConsumerDialog.class;
   JmsConsumerMeta jmsMeta;
   private DestinationForm destinationForm;
   private ConnectionForm connectionForm;

plugins/streaming/impls/jms/src/main/resources/OSGI-INF/blueprint/blueprint.xml

@@ -25,16 +25,16 @@
 
 
   <bean id="JmsConsumer" class="org.pentaho.di.trans.step.jms.JmsConsumerMeta" scope="prototype">
-    <argument ref="jmsContextProviders"/>
+    <argument ref="jmsProviders"/>
     <pen:di-plugin type="org.pentaho.di.core.plugins.StepPluginType"/>
   </bean>
 
 
-  <reference-list id="jmsContextProviders" interface="org.pentaho.di.trans.step.jms.context.JmsContextProvider"
+  <reference-list id="jmsProviders" interface="org.pentaho.di.trans.step.jms.context.JmsProvider"
                   availability="optional"/>
 
-  <service interface="org.pentaho.di.trans.step.jms.context.JmsContextProvider">
-    <bean class="org.pentaho.di.trans.step.jms.context.WebsphereMQContextProvider" />
+  <service interface="org.pentaho.di.trans.step.jms.context.JmsProvider">
+    <bean class="org.pentaho.di.trans.step.jms.context.WebsphereMQProvider" />
   </service>
 
 </blueprint>

plugins/streaming/impls/jms/src/main/resources/org/pentaho/di/trans/step/jms/ui/messages/messages_en_US.properties → plugins/streaming/impls/jms/src/main/resources/org/pentaho/di/trans/step/jms/messages/messages_en_US.properties

@@ -9,3 +9,5 @@ JmsDialog.DestinationType=Destination Type
 JmsDialog.Dest.Topic=Topic
 JmsDialog.Dest.Queue=Queue
 JmsDialog.Dest.Name=Destination name:
+
+JmsWebsphereMQ.DestinationNameRequired=Destination name must be set.

plugins/streaming/impls/jms/src/test/java/org/pentaho/di/trans/step/jms/JmsStreamSourceTest.java

@@ -37,10 +37,8 @@
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageListener;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.Objects;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.junit.Assert.assertThat;
@@ -86,16 +84,4 @@ public void testReceiveMessage() {
   }
 
 
-  List<String> foo( String s ) {
-    return Arrays.asList( null, "foo" );
-  }
-
-  @Test public void test() {
-    System.out.println( foo( "fo" )
-      .stream()
-      .filter( Objects::nonNull )
-      .findFirst().get() );
-
-
-  }
 }