Refactor service definition
package org.apache.streampipes.sinks.brokers.jvm;
-import org.apache.streampipes.container.init.DeclarersSingleton;
+import org.apache.streampipes.container.model.SpServiceDefinition;
+import org.apache.streampipes.container.model.SpServiceDefinitionBuilder;
import org.apache.streampipes.container.standalone.init.StandaloneModelSubmitter;
import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
@@ -28,10 +29,8 @@ import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
import org.apache.streampipes.sinks.brokers.jvm.bufferrest.BufferRestController;
-import org.apache.streampipes.sinks.brokers.jvm.config.BrokersJvmConfig;
import org.apache.streampipes.sinks.brokers.jvm.jms.JmsController;
import org.apache.streampipes.sinks.brokers.jvm.kafka.KafkaController;
-import org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttPublisherSink;
import org.apache.streampipes.sinks.brokers.jvm.nats.NatsController;
import org.apache.streampipes.sinks.brokers.jvm.pulsar.PulsarController;
import org.apache.streampipes.sinks.brokers.jvm.rabbitmq.RabbitMqController;
@@ -41,28 +40,25 @@ import org.apache.streampipes.sinks.brokers.jvm.websocket.WebsocketServerSink;
public class BrokersJvmInit extends StandaloneModelSubmitter {
public static void main(String[] args) {
- DeclarersSingleton.getInstance()
- .add(new KafkaController())
- .add(new JmsController())
- .add(new RestController())
- .add(new BufferRestController())
- .add(new RabbitMqController())
- .add(new MqttPublisherSink())
- .add(new WebsocketServerSink())
- .add(new PulsarController())
- .add(new NatsController());
- DeclarersSingleton.getInstance().registerDataFormats(
- new JsonDataFormatFactory(),
+ SpServiceDefinition serviceDef = SpServiceDefinitionBuilder.create("org.apache.streampipes.sinks.notifications.jvm", "Sinks Notifications JVM", "", 8096)
+ .registerPipelineElements(new KafkaController(),
+ new JmsController(),
+ new RestController(),
+ new BufferRestController(),
+ new RabbitMqController(),
+ new WebsocketServerSink(),
+ new PulsarController(),
+ new NatsController())
+ .registerMessagingFormats(new JsonDataFormatFactory(),
new CborDataFormatFactory(),
new SmileDataFormatFactory(),
- new FstDataFormatFactory());
+ new FstDataFormatFactory())
+ .registerMessagingProtocols(new SpKafkaProtocolFactory(),
+ new SpJmsProtocolFactory(),
+ new SpMqttProtocolFactory())
+ .build();
- DeclarersSingleton.getInstance().registerProtocols(
- new SpKafkaProtocolFactory(),
- new SpMqttProtocolFactory(),
- new SpJmsProtocolFactory());
-
- new BrokersJvmInit().init(BrokersJvmConfig.INSTANCE);
+ new BrokersJvmInit().init(serviceDef);
}
}
