@Test(expected = Exception.class) public void testInjectionExceptionOfSink() throws IOException { final Map<String, String> conf = TextSocketSourceConfiguration.newBuilder() .setHostAddress("localhost") .setHostPort(13666) .build().getConfiguration(); generator.newSink(conf, classLoader); }
@SuppressWarnings("unchecked") public <T> Sink<T> newSink( final Map<String, String> conf, final ClassLoader classLoader) throws IOException { final String type = conf.get(ConfKeys.SinkConf.SINK_TYPE.name()); if (type.equals(ConfValues.SinkType.NETTY.name())) { final String serverAddress = conf.get(ConfKeys.NettySink.SINK_ADDRESS.name()); final int serverPort = Integer.valueOf(conf.get(ConfKeys.NettySink.SINK_PORT.name())); return (Sink<T>)new NettyTextSink(serverAddress, serverPort, nettySharedResource, identifierFactory); } else if (type.equals(ConfValues.SinkType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MqttSink.MQTT_SINK_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MqttSink.MQTT_SINK_TOPIC.name()); try { return (Sink<T>)new MqttSink(brokerURI, topic, mqttSharedResource); } catch (final MqttException e) { e.printStackTrace(); throw new IOException(e); } } else { throw new RuntimeException("Invalid sink type: " + type); } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public <T> Sink<T> newSink( final Map<String, String> conf, final ClassLoader classLoader) throws IOException { final String type = conf.get(ConfKeys.SinkConf.SINK_TYPE.name()); if (type.equals(ConfValues.SinkType.NETTY.name())) { final String serverAddress = conf.get(ConfKeys.NettySink.SINK_ADDRESS.name()); final int serverPort = Integer.valueOf(conf.get(ConfKeys.NettySink.SINK_PORT.name())); return (Sink<T>)new NettyTextSink(serverAddress, serverPort, nettySharedResource, identifierFactory); } else if (type.equals(ConfValues.SinkType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MqttSink.MQTT_SINK_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MqttSink.MQTT_SINK_TOPIC.name()); try { return (Sink<T>)new MqttSink(brokerURI, topic, mqttSharedResource); } catch (final MqttException e) { e.printStackTrace(); throw new IOException(e); } } else { throw new RuntimeException("Invalid sink type: " + type); } } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public <T> Sink<T> newSink( final Map<String, String> conf, final ClassLoader classLoader) throws IOException { final String type = conf.get(ConfKeys.SinkConf.SINK_TYPE.name()); if (type.equals(ConfValues.SinkType.NETTY.name())) { final String serverAddress = conf.get(ConfKeys.NettySink.SINK_ADDRESS.name()); final int serverPort = Integer.valueOf(conf.get(ConfKeys.NettySink.SINK_PORT.name())); return (Sink<T>)new NettyTextSink(serverAddress, serverPort, nettySharedResource, identifierFactory); } else if (type.equals(ConfValues.SinkType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MqttSink.MQTT_SINK_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MqttSink.MQTT_SINK_TOPIC.name()); try { return (Sink<T>)new MqttSink(brokerURI, topic, mqttSharedResource); } catch (final MqttException e) { e.printStackTrace(); throw new IOException(e); } } else { throw new RuntimeException("Invalid sink 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 <T> Sink<T> newSink( final Map<String, String> conf, final ClassLoader classLoader) throws IOException { final String type = conf.get(ConfKeys.SinkConf.SINK_TYPE.name()); if (type.equals(ConfValues.SinkType.NETTY.name())) { final String serverAddress = conf.get(ConfKeys.NettySink.SINK_ADDRESS.name()); final int serverPort = Integer.valueOf(conf.get(ConfKeys.NettySink.SINK_PORT.name())); return (Sink<T>)new NettyTextSink(serverAddress, serverPort, nettySharedResource, identifierFactory); } else if (type.equals(ConfValues.SinkType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MqttSink.MQTT_SINK_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MqttSink.MQTT_SINK_TOPIC.name()); try { return (Sink<T>)new MqttSink(brokerURI, topic, mqttSharedResource); } catch (final MqttException e) { e.printStackTrace(); throw new IOException(e); } } else { throw new RuntimeException("Invalid sink 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 <T> Sink<T> newSink( final Map<String, String> conf, final ClassLoader classLoader) throws IOException { final String type = conf.get(ConfKeys.SinkConf.SINK_TYPE.name()); if (type.equals(ConfValues.SinkType.NETTY.name())) { final String serverAddress = conf.get(ConfKeys.NettySink.SINK_ADDRESS.name()); final int serverPort = Integer.valueOf(conf.get(ConfKeys.NettySink.SINK_PORT.name())); return (Sink<T>)new NettyTextSink(serverAddress, serverPort, nettySharedResource, identifierFactory); } else if (type.equals(ConfValues.SinkType.MQTT.name())) { final String brokerURI = conf.get(ConfKeys.MqttSink.MQTT_SINK_BROKER_URI.name()); final String topic = conf.get(ConfKeys.MqttSink.MQTT_SINK_TOPIC.name()); try { return (Sink<T>)new MqttSink(brokerURI, topic, mqttSharedResource); } catch (final MqttException e) { e.printStackTrace(); throw new IOException(e); } } else { throw new RuntimeException("Invalid sink 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();  }