@Test public void testSuccessOfMQTTDataGenerator() throws Exception { final Map<String, String> conf = MQTTSourceConfiguration.newBuilder() .setBrokerURI("tcp: .setTopic("topic") .build().getConfiguration(); final DataGenerator<MqttMessage> dataGenerator = generator.newDataGenerator(conf, classLoader); Assert.assertTrue(dataGenerator instanceof MQTTDataGenerator); }
@SuppressWarnings("unchecked") public DataGenerator newDataGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.SourceConf.SOURCE_TYPE.name()); if (type.equals(ConfValues.SourceType.KAFKA.name())) { final String topic = conf.get(ConfKeys.KafkaSourceConf.KAFKA_TOPIC.name()); final Map<String, Object> kafkaConsumerConf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.KafkaSourceConf.KAFKA_CONSUMER_CONFIG.name()), classLoader); return new KafkaDataGenerator(topic, kafkaConsumerConf, kafkaSharedResource); } else if (type.equals(ConfValues.SourceType.NETTY.name())) { final String addr = conf.get(ConfKeys.NettySourceConf.SOURCE_ADDR.name()); final int port = Integer.valueOf(conf.get(ConfKeys.NettySourceConf.SOURCE_PORT.name())); return new NettyTextDataGenerator(addr, port, nettySharedResource); } else if (type.equals(ConfValues.SourceType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_TOPIC.name()); return mqttSharedResource.getDataGenerator(brokerURI, topic); } else { throw new RuntimeException("Invalid source type: " + type); } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public DataGenerator newDataGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.SourceConf.SOURCE_TYPE.name()); if (type.equals(ConfValues.SourceType.KAFKA.name())) { final String topic = conf.get(ConfKeys.KafkaSourceConf.KAFKA_TOPIC.name()); final Map<String, Object> kafkaConsumerConf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.KafkaSourceConf.KAFKA_CONSUMER_CONFIG.name()), classLoader); return new KafkaDataGenerator(topic, kafkaConsumerConf, kafkaSharedResource); } else if (type.equals(ConfValues.SourceType.NETTY.name())) { final String addr = conf.get(ConfKeys.NettySourceConf.SOURCE_ADDR.name()); final int port = Integer.valueOf(conf.get(ConfKeys.NettySourceConf.SOURCE_PORT.name())); return new NettyTextDataGenerator(addr, port, nettySharedResource); } else if (type.equals(ConfValues.SourceType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_TOPIC.name()); return mqttSharedResource.getDataGenerator(brokerURI, topic); } else { throw new RuntimeException("Invalid source type: " + type); } } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public DataGenerator newDataGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.SourceConf.SOURCE_TYPE.name()); if (type.equals(ConfValues.SourceType.KAFKA.name())) { final String topic = conf.get(ConfKeys.KafkaSourceConf.KAFKA_TOPIC.name()); final Map<String, Object> kafkaConsumerConf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.KafkaSourceConf.KAFKA_CONSUMER_CONFIG.name()), classLoader); return new KafkaDataGenerator(topic, kafkaConsumerConf, kafkaSharedResource); } else if (type.equals(ConfValues.SourceType.NETTY.name())) { final String addr = conf.get(ConfKeys.NettySourceConf.SOURCE_ADDR.name()); final int port = Integer.valueOf(conf.get(ConfKeys.NettySourceConf.SOURCE_PORT.name())); return new NettyTextDataGenerator(addr, port, nettySharedResource); } else if (type.equals(ConfValues.SourceType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_TOPIC.name()); return mqttSharedResource.getDataGenerator(brokerURI, topic); } else { throw new RuntimeException("Invalid source type: " + type); } } @Inject private  PhysicalObjectGenerator(final ScheduledExecutorServiceWrapper schedulerWrapper,
                                  final KafkaSharedResource kafkaSharedResource,
                                  final NettySharedResource nettySharedResource,
                                  final MQTTResource mqttSharedResource,
                                  @Parameter(PeriodicCheckpointPeriod.class) final long checkpointPeriod,
                                  final StringIdentifierFactory identifierFactory); }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public DataGenerator newDataGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.SourceConf.SOURCE_TYPE.name()); if (type.equals(ConfValues.SourceType.KAFKA.name())) { final String topic = conf.get(ConfKeys.KafkaSourceConf.KAFKA_TOPIC.name()); final Map<String, Object> kafkaConsumerConf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.KafkaSourceConf.KAFKA_CONSUMER_CONFIG.name()), classLoader); return new KafkaDataGenerator(topic, kafkaConsumerConf, kafkaSharedResource); } else if (type.equals(ConfValues.SourceType.NETTY.name())) { final String addr = conf.get(ConfKeys.NettySourceConf.SOURCE_ADDR.name()); final int port = Integer.valueOf(conf.get(ConfKeys.NettySourceConf.SOURCE_PORT.name())); return new NettyTextDataGenerator(addr, port, nettySharedResource); } else if (type.equals(ConfValues.SourceType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_TOPIC.name()); return mqttSharedResource.getDataGenerator(brokerURI, topic); } else { throw new RuntimeException("Invalid source type: " + type); } } @Inject private  PhysicalObjectGenerator(final ScheduledExecutorServiceWrapper schedulerWrapper,
                                  final KafkaSharedResource kafkaSharedResource,
                                  final NettySharedResource nettySharedResource,
                                  final MQTTResource mqttSharedResource,
                                  @Parameter(PeriodicCheckpointPeriod.class) final long checkpointPeriod,
                                  final StringIdentifierFactory identifierFactory); @SuppressWarnings("unchecked") EventGenerator<T> newEventGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") DataGenerator newDataGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Operator newOperator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Sink<T> newSink(
      final Map<String, String> conf,
      final ClassLoader classLoader); @Override void close(); }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public DataGenerator newDataGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.SourceConf.SOURCE_TYPE.name()); if (type.equals(ConfValues.SourceType.KAFKA.name())) { final String topic = conf.get(ConfKeys.KafkaSourceConf.KAFKA_TOPIC.name()); final Map<String, Object> kafkaConsumerConf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.KafkaSourceConf.KAFKA_CONSUMER_CONFIG.name()), classLoader); return new KafkaDataGenerator(topic, kafkaConsumerConf, kafkaSharedResource); } else if (type.equals(ConfValues.SourceType.NETTY.name())) { final String addr = conf.get(ConfKeys.NettySourceConf.SOURCE_ADDR.name()); final int port = Integer.valueOf(conf.get(ConfKeys.NettySourceConf.SOURCE_PORT.name())); return new NettyTextDataGenerator(addr, port, nettySharedResource); } else if (type.equals(ConfValues.SourceType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MQTTSourceConf.MQTT_SRC_TOPIC.name()); return mqttSharedResource.getDataGenerator(brokerURI, topic); } else { throw new RuntimeException("Invalid source type: " + type); } } @Inject private  PhysicalObjectGenerator(final ScheduledExecutorServiceWrapper schedulerWrapper,
                                  final KafkaSharedResource kafkaSharedResource,
                                  final NettySharedResource nettySharedResource,
                                  final MQTTResource mqttSharedResource,
                                  @Parameter(PeriodicCheckpointPeriod.class) final long checkpointPeriod,
                                  final StringIdentifierFactory identifierFactory); @SuppressWarnings("unchecked") EventGenerator<T> newEventGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") DataGenerator newDataGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Operator newOperator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Sink<T> newSink(
      final Map<String, String> conf,
      final ClassLoader classLoader); @Override void close();  }