init skeleton for centralized app

src/main/java/ipos/project/SensorValueIntegration/api/mqtt/impl/MqttHandlerImpl.java

+package ipos.project.SensorValueIntegration.api.mqtt.impl;
+
+import ipos.project.DataModellntegration.api.mqtt.Handler;
+import ipos.project.DataModellntegration.api.mqtt.MqttListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class MqttHandlerImpl {
+
+    private final ApplicationContext ctx;
+    private final Map<String, Handler> handlerMap = new HashMap<>();
+    
+    @Autowired
+    public MqttHandlerImpl(ApplicationContext ctx) {
+        this.ctx = ctx;
+    }
+
+    public Handler getMessageHandler(String topic) {
+        if (handlerMap.isEmpty()) {
+            // Find beans with @MqttHandler annotation
+            Collection<Object> containers = ctx.getBeansWithAnnotation(MqttListener.class).values();
+            if (containers.size()>0) {
+                for (Object container : containers) {
+                    String containerTopic = container.getClass().getAnnotation(MqttListener.class).value();
+                    handlerMap.put(containerTopic, (Handler)container);
+                }
+            }
+        }
+        return handlerMap.get(topic);
+    }
+
+}

src/main/java/ipos/project/SensorValueIntegration/config/MqttConfig.java

+package ipos.project.SensorValueIntegration.config;
+
+import ipos.project.SensorValueIntegration.api.mqtt.impl.MqttHandlerImpl;
+import org.eclipse.paho.client.mqttv3.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.sql.Timestamp;
+
+@Configuration
+public class MqttConfig {
+
+    private final Logger LOG = LoggerFactory.getLogger(getClass());
+    private final MqttHandlerImpl messageHandler;
+
+    @Autowired
+    public MqttConfig(MqttHandlerImpl mqttHandler) {
+        this.messageHandler = mqttHandler;
+    }
+
+    @Bean
+    @ConfigurationProperties(prefix = "mqtt")
+    public MqttConnectOptions mqttConnectOptions() {
+        return new MqttConnectOptions();
+    }
+
+    @Bean
+    public MqttAsyncClient mqttClient(@Value("${mqtt.clientId}") String clientId,
+                                      @Value("${mqtt.hostname}") String hostname,
+                                      @Value("${mqtt.port}") int port) throws MqttException {
+
+        LOG.info("MQTT host: tcp://" + hostname + ":" + port);
+
+        MqttAsyncClient mqttClient = new MqttAsyncClient("tcp://" + hostname + ":" + port, clientId);
+
+        mqttClient.setCallback(new MqttCallback() {
+            public void messageArrived(String topic, MqttMessage message) throws Exception {
+                String time = new Timestamp(System.currentTimeMillis()).toString();
+                LOG.debug("\nReceived a Message: " +
+                        "\n\tTime:    " + time +
+                        "\n\tTopic:   " + topic +
+                        "\n\tMessage: " + new String(message.getPayload()) +
+                        "\n\tQoS:     " + message.getQos() + "\n");
+                messageHandler.getMessageHandler(topic).handle(message);
+            }
+
+            public void connectionLost(Throwable cause) {
+                LOG.warn("Connection to MQTT broker lost: " + cause.getMessage());
+                cause.printStackTrace();
+            }
+
+            public void deliveryComplete(IMqttDeliveryToken token) {
+            }
+
+        });
+
+        mqttClient.connect(mqttConnectOptions());
+
+        return mqttClient;
+    }
+
+}

src/main/java/ipos/project/SensorValueIntegration/service/ExternalCommunicationService.java

+package ipos.project.SensorValueIntegration.service;
+
+public interface ExternalCommunicationService {
+    void publish(final String topic, final Object objJson , int qos, boolean retained);
+    void subscribe(final String topic, int qos);
+}

src/main/java/ipos/project/SensorValueIntegration/service/impl/ExtMqttServiceImpl.java

+package ipos.project.SensorValueIntegration.service.impl;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import ipos.project.SensorValueIntegration.service.ExternalCommunicationService;
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Service
+public class ExtMqttServiceImpl implements ExternalCommunicationService {
+
+    private final Logger LOG = LoggerFactory.getLogger(getClass());
+
+    private final MqttAsyncClient mqttClient;
+    ObjectMapper mapper;
+
+    @Autowired
+    public ExtMqttServiceImpl(MqttAsyncClient mqttClient) {
+        this.mqttClient = mqttClient;
+    }
+
+    //                                                                                                   ====== Publish
+    //                                                                                          --- Client Json objects
+    public void publish(final String topic, final Object objJson , int qos, boolean retained) {
+        MqttMessage mqttMessage = new MqttMessage();
+        mqttMessage.setPayload(objJson.toString().getBytes());
+        mqttMessage.setQos(qos);
+        mqttMessage.setRetained(retained);
+
+        try {
+            mqttClient.publish(topic, mqttMessage);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+
+    //                                                                                                 ====== Subscribe
+    public void subscribe(final String topic, int qos) {
+        LOG.info(">> Subscribe on " + topic);
+        try {
+            mqttClient.subscribe(topic, qos);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/ipos/project/UseCaseController/PositionMonitoring.java

+package ipos.project.UseCaseController;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.jms.annotation.JmsListener;
+import org.springframework.stereotype.Component;
+import ipos.models.SimpleScene.*;
+
+
+@Component
+public class PositionMonitoring {
+    private final Logger LOG = LoggerFactory.getLogger(getClass());
+
+    @JmsListener(destination = "/positions123", containerFactory = "myFactory")
+    public void receiveMessage(IposPosition pos) {
+        LOG.info("Received <" + pos + ">");
+    }
+}

src/main/java/ipos/project/config/JmsConfig.java

+package ipos.project.config;
+
+import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
+import org.springframework.jms.config.JmsListenerContainerFactory;
+import org.springframework.jms.support.converter.MarshallingMessageConverter;
+import org.springframework.jms.support.converter.MessageConverter;
+import org.springframework.jms.support.converter.MessageType;
+
+import javax.jms.ConnectionFactory;
+
+@Configuration
+public class JmsConfig {
+
+    @Bean
+    public JmsListenerContainerFactory<?> myFactory(ConnectionFactory connectionFactory,
+                                                    DefaultJmsListenerContainerFactoryConfigurer configurer) {
+        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
+        // This provides all boot's default to this factory, including the message converter
+        configurer.configure(factory, connectionFactory);
+        // You could still override some of Boot's default if necessary.
+        return factory;
+    }
+
+    @Bean // Serialize message content to json using TextMessage
+    public MessageConverter jacksonJmsMessageConverter() {
+        MarshallingMessageConverter converter = new MarshallingMessageConverter();
+        converter.setTargetType(MessageType.BYTES);
+        return converter;
+    }
+}

src/main/resources/SimpleScene.proto

+syntax = "proto3";
+
+package ipos.models;
+
+message IposConfig {
+  repeated IposFrame frames = 1;
+  repeated IposObject objects = 2;
+}
+
+message IposPositionUpdate {
+  repeated IposObject objects = 1;
+}
+
+message IposObject {
+
+  // required:
+  string id = 1; // unique for each object, but there may be multiple sensors
+  string sensor_id = 2; // unique
+  string type = 3; // Applications may send BOX,BIN,ZONE,ROBOT,HUMAN,OTHER
+
+  // optional:
+  IposRelativePosition position = 4;
+  IposSimpleOrientation orientation = 5;
+  string last_pos_update = 6; // https://en.wikipedia.org/wiki/ISO_8601
+
+}
+
+message IposFrame {
+  string id = 1; // unique for every frame, used for relative coordinates
+  IposSpace space = 2; // cube defining the extension, might be (0,0,0)
+  IposPosition position = 3; // centre of the "frame" cube, absolute
+  IposSimpleOrientation orientation = 4; // orientation of the frame
+  float delta = 5; // min. Position to trigger a position update
+}
+
+// relative coordinates
+message IposRelativePosition {
+  string frame_id = 1;
+  IposPosition pos = 2;
+}
+
+// relative coordinates
+message IposPosition {
+  float x = 1; // pos in x direction in m
+  float y = 2; // pos in y direction in m
+  float z = 3; // pos in z direction in m (up)
+  float accuracy = 4; // object is with 95% probability within x metres of this point (gaussian distribution)
+}
+
+message IposSpace {
+  float x = 1; // size in x direction in m
+  float y = 2; // size in y direction in m
+  float z = 3; // size in z direction in m (height)
+}
+
+message IposSimpleOrientation {
+  float x = 1; // quaternion notation
+  float y = 2; // if the quaternion is (0,0,0,0)
+  float z = 3; // there is no orientation
+  float w = 4; // available
+}
+
+message IposSubscriptionRequest {
+
+  string subscription_id = 1;
+
+  Selection type_selection = 2;
+  repeated string types = 3;
+
+  Selection object_selection = 4;
+  repeated string objects = 5;
+
+  Selection zone_selection = 6;
+  repeated string zones = 7;
+
+  double minimum_position_delta = 8;
+  double max_update_frequency = 9;
+
+}
+
+enum Selection {
+  SELECTION_UNSPECIFIED = 0;
+  SELECTION_ALL = 1;
+  SELECTION_NONE = 2;
+  SELECTION_ONLY_LIST = 3;
+  SELECTION_EXCEPT_LIST = 4;
+}

src/main/resources/application.yml

+

src/test/java/ipos/project/MainAppTests.java

+package ipos.project;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class MainAppTests {
+
+	@Test
+	void contextLoads() {
+	}
+
+}