Made broker and topic dynamic
@@ -128,11 +128,14 @@ public class SpConnect extends AbstractRestInterface {
} else {
ds = ((AdapterStreamDescription) adapterDescription).getDataStream();
- //TODO just for debugging remvove
- TransportProtocol tp = Protocols.kafka("141.21.42.75", 9092, "test.topic");
+ String topic = getTopicPrefix() + adapterDescription.getName();
+
+ TransportProtocol tp = Protocols.kafka(BackendConfig.INSTANCE.getKafkaHost(), BackendConfig.INSTANCE.getKafkaPort(), topic);
EventGrounding eg = new EventGrounding();
eg.setTransportFormats(Arrays.asList(Formats.jsonFormat()));
eg.setTransportProtocol(tp);
+
+ //TODO remove when frontend schema fixed
EventSchema es = new EventSchema();
es.setEventProperties(Arrays.asList(EpProperties.timestampProperty("timestamp")));
ds.setEventSchema(es);
@@ -254,8 +257,9 @@ public class SpConnect extends AbstractRestInterface {
private void startStreamAdapter(AdapterStreamDescription asd) {
- String brokerUrl = "141.21.42.75:9092";
- String topic = "test.topic";
+ String brokerUrl = BackendConfig.INSTANCE.getKafkaUrl();
+ String topic = getTopicPrefix() + asd.getName();
+
Adapter adapter = new Adapter(brokerUrl, topic, false);
// RunningAdapterInstances.INSTANCE.addAdapter(dataSet.getDatasetInvocationId(), adapter);
// TODO execute a Thread
@@ -291,5 +295,8 @@ public class SpConnect extends AbstractRestInterface {
return ok(JsonLdUtils.toJsonLD(adapterDescriptionList));
}
+ private String getTopicPrefix() {
+ return "org.streampipes.autogenerated.adapters.";
+ }
}
