Update generic-storage-agent
Define the StorageAgent as abstract and implementing the AgentRelay interface
Define the history keys property allowing to define the attributes path holding values to be stored in history
Update influxdb-storage-agent and http-storage-agent modules based on the updated generic one
diff --git a/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/Stack.java b/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/Stack.java
index 59aaa48..f59a1a2 100644
--- a/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/Stack.java
+++ b/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/Stack.java
@@ -32,7 +32,7 @@
public void push(JSONObject element) {
if (element == null)
- throw new IllegalArgumentException("Can't Push a null element");
+ return;
synchronized (list) {
list.add(element);
}
diff --git a/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageAgent.java b/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageAgent.java
index fca4d93..3b704ef 100644
--- a/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageAgent.java
+++ b/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageAgent.java
@@ -10,122 +10,256 @@
*/
package org.eclipse.sensinact.gateway.agent.storage.generic;
-import java.io.IOException;
import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.Date;
+import java.util.Dictionary;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
-import org.eclipse.sensinact.gateway.common.execution.DefaultErrorHandler;
-import org.eclipse.sensinact.gateway.common.execution.ErrorHandler;
+import org.eclipse.sensinact.gateway.common.execution.Executable;
import org.eclipse.sensinact.gateway.core.DataResource;
import org.eclipse.sensinact.gateway.core.LocationResource;
import org.eclipse.sensinact.gateway.core.Resource;
-import org.eclipse.sensinact.gateway.core.message.AbstractMidAgentCallback;
-import org.eclipse.sensinact.gateway.core.message.SnaErrorMessageImpl;
import org.eclipse.sensinact.gateway.core.message.SnaLifecycleMessage.Lifecycle;
import org.eclipse.sensinact.gateway.core.message.SnaLifecycleMessageImpl;
-import org.eclipse.sensinact.gateway.core.message.SnaResponseMessage;
+import org.eclipse.sensinact.gateway.core.message.SnaMessage;
import org.eclipse.sensinact.gateway.core.message.SnaUpdateMessageImpl;
+import org.eclipse.sensinact.gateway.core.message.whiteboard.AbstractAgentRelay;
import org.eclipse.sensinact.gateway.util.UriUtils;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
+ * {@link AgentRelay} in charge of relaying event notifications to the {@link StorageConnection}
+ * in charge of the effective data storage
+ *
* @author <a href="mailto:christophe.munilla@cea.fr">Christophe Munilla</a>
*/
-public class StorageAgent extends AbstractMidAgentCallback {
+public abstract class StorageAgent extends AbstractAgentRelay {
+
private static final Logger LOG = LoggerFactory.getLogger(StorageAgent.class);
+
+ protected static final String STORAGE_AGENT_KEYS_PROPS = "org.eclipse.sensinact.gateway.history.keys";
+
private static final SimpleDateFormat FORMAT = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
- private final StorageConnection storageConnection;
+
+
+ private Map<String,Executable<SnaMessage<?>,String>> keyProcessors;
+ private Map<String,Object> storageKeyValuesMap;
+ private Map<String,String> storageKeyNamesMap;
+
+ /**
+ * The string formated location of service providers that have already been
+ * processed by this {@link StorageAgent}
+ */
+ private Map<String, String> locations;
+
+ /**
+ * the {@link StorageConnection} in charge of effective storage
+ */
+ private StorageConnection storageConnection;
/**
* Constructor
- *
- * @param storageConnection
- * @throws IOException Exception on connection problem
*/
- public StorageAgent(StorageConnection storageConnection) throws IOException {
+ public StorageAgent() {
super();
- this.storageConnection = storageConnection;
- super.setErrorHandler(new DefaultErrorHandler(ErrorHandler.Policy.LOG|ErrorHandler.Policy.IGNORE));
+ this.locations = new HashMap<>();
+ this.storageKeyValuesMap = new HashMap<>();
+ this.keyProcessors = new HashMap<>();
+// Key processor example
+ this.keyProcessors.put("path", new Executable<SnaMessage<?>,String>(){
+ @Override
+ public String execute(SnaMessage<?> message) throws Exception {
+ return message.getPath();
+ }
+ });
+ this.keyProcessors.put("location", new Executable<SnaMessage<?>,String>(){
+ @Override
+ public String execute(SnaMessage<?> message) throws Exception {
+ String uri = message.getPath();
+ String[] pathElements = UriUtils.getUriElements(uri);
+ return StorageAgent.this.getLocation(pathElements[0]);
+ }
+ });
+// Define keys mapping at initialization time
}
+ /**
+ * Defines the mapping between key names and the String paths of the attributes from which the mapped values
+ * will be extracted - ex : <key>=/<service>/<resource>/<attribute> or <key>=/<service>/<resource>, in this last
+ * case the default 'value' attribute will be used
+ *
+ * @param keys colon separated String formated keys mapping
+ */
+ protected void setStorageKeys(String keys) {
+ if(keys!=null)
+ this.storageKeyNamesMap = Arrays.asList(keys.split(",")).stream().<HashMap<String,String>>collect(
+ HashMap::new,(h,prop)-> {
+ String keyValuePair[] = prop.split("=");
+ String key = keyValuePair[0].trim();
+ if(UriUtils.getUriElements(key).length == 2)
+ key = key.concat("/value");
+ h.put(key, keyValuePair[1].trim());
+ }, Map::putAll);
+ else
+ this.storageKeyNamesMap = Collections.<String,String>emptyMap();
+ }
+
+ /**
+ * Registers a new key processor, allowing to feed the data object that will be stored
+ *
+ * @param key the key String name
+ * @param executor the {@link Executable} allowing to process an SnaMessage to extract
+ * the value to be mapped to the specified key
+ */
+ public void addFixKeyProcessor(String key, Executable<SnaMessage<?>,String> executor) {
+ if(key !=null && executor !=null)
+ this.keyProcessors.put(key, executor);
+ }
+
+ /**
+ * Sets the {@link StorageConnection} in charge of the effective data storage
+ *
+ * @param storageConnection the {@link StorageConnection} in charge of the effective data storage
+ */
+ protected void setStorageConnection(StorageConnection storageConnection) {
+ this.storageConnection = storageConnection;
+ }
+
+ /**
+ * Returns the String location for the service provider whose path is passed as
+ * parameter
+ *
+ * @param path
+ * the path of the service provider for which to retrieve the string
+ * location
+ * @return the String location for the specified path
+ */
+ protected String getLocation(String serviceProvider) {
+ synchronized (this.locations) {
+ return this.locations.get(serviceProvider);
+ }
+ }
+
+ /**
+ * Sets the String location for the service provider whose path is passed as
+ * parameter
+ *
+ * @param path
+ * the path of the service provider for which to set the string
+ * location
+ * @param location
+ * the string location to set
+ */
+ protected void setLocation(String serviceProvider, String location) {
+ synchronized (this.locations) {
+ this.locations.put(serviceProvider, location);
+ }
+ }
+
+ private Dictionary<String,Object> preProcessSnaMessage(SnaMessage<?> message){
+ final Dictionary<String,Object> ts = new Hashtable<>();
+
+ for(Iterator<String> it = this.keyProcessors.keySet().iterator();it.hasNext();) {
+ String key = it.next();
+ String val = null;
+ try {
+ val = this.keyProcessors.get(key).execute(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ if(val != null)
+ ts.put(key,val);
+ }
+ return ts;
+ }
@Override
public void doHandle(SnaUpdateMessageImpl message) {
- String path = message.getPath();
- LOG.debug("storage agent informed of an update on {}...", path);
- String[] elements = UriUtils.getUriElements(path);
- String serviceProvider = elements[0];
- String service = elements[1];
- String resource = elements[2];
- JSONObject initial = message.getNotification();
- this.doHandle(serviceProvider, service, resource, initial);
+ this.doHandle( message.getPath(), message.getNotification(),preProcessSnaMessage(message));
}
@Override
public void doHandle(SnaLifecycleMessageImpl message) {
- String path = message.getPath();
- if (!Lifecycle.RESOURCE_APPEARING.equals(message.getType()) || Resource.Type.ACTION.equals(message.getNotification(Resource.Type.class, "type"))) {
+ if (!Lifecycle.RESOURCE_APPEARING.equals(message.getType()) || Resource.Type.ACTION.equals(message.getNotification(Resource.Type.class, "type")))
return;
- }
- String[] elements = UriUtils.getUriElements(path);
- String serviceProvider = elements[0];
- String service = elements[1];
- String resource = elements[2];
- JSONObject initial = new JSONObject(message.getJSON()).getJSONObject("initial");
- this.doHandle(serviceProvider, service, resource, initial);
+ this.doHandle( message.getPath(), message.<JSONObject>get("initial"), preProcessSnaMessage(message));
}
- /**
- * @param serviceProvider the service provider
- * @param service the service
- * @param resource the resource
- * @param content the content
- */
- private void doHandle(String serviceProvider, String service, String resource, JSONObject content) {
- Object initialValue = content.opt(DataResource.VALUE);
- if (JSONObject.NULL.equals(initialValue)) {
+ //
+ private void doHandle(String path, JSONObject content, Dictionary<String, Object> ts) {
+ Object value = content.opt(DataResource.VALUE);
+ if (JSONObject.NULL.equals(value)) {
//exclude initial null value
- LOG.debug("Unexpected null initial value error {}/{}/{}/{}...", serviceProvider, service, resource, initialValue);
return;
- }
- Long timestamp = (Long) content.opt("timestamp");
- if (timestamp == null) {
- timestamp = System.currentTimeMillis();
- }
- String timestampStr = FORMAT.format(new Date(timestamp));
+ }
+ String[] pathElements = UriUtils.getUriElements(path);
+ String provider = pathElements[0];
+ String resource = pathElements[2];
+
if (LocationResource.LOCATION.equalsIgnoreCase(resource)) {
//set location and escape
- try {
- super.setLocation(serviceProvider, String.valueOf(initialValue));
- } catch (Exception e) {
- LOG.debug("Unexpected location error {}", e.getMessage(), e);
- }
+ this.setLocation(provider, String.valueOf(value));
+ return;
}
+ if(this.storageConnection == null)
+ return;
+
+ String attribute = (String) content.opt("name");
+
+ if(pathElements[2].equals(attribute))
+ attribute = "value";
+
+ if(this.storageKeyNamesMap!=null) {
+
+ Set<String> keys = this.storageKeyNamesMap.keySet();
+ String serviceUri = UriUtils.getUri(new String[] {pathElements[1],pathElements[2],attribute});
+
+ if(keys.contains(serviceUri)) {
+
+ this.storageKeyValuesMap.put(UriUtils.getUri(new String[] {pathElements[0], pathElements[1], pathElements[2], attribute}),value);
+ return;
+
+ } else {
+
+ final String serviceProviderId = pathElements[0];
+ keys.forEach(s -> {
+ String p = UriUtils.getUri(new String[] {serviceProviderId, s});
+ Object tagValue = storageKeyValuesMap.get(p);
+ if(tagValue != null)
+ ts.put(this.storageKeyNamesMap.get(s), tagValue);
+ });
+ }
+ }
+ Long timestamp = (Long) content.opt("timestamp");
+ if (timestamp == null)
+ timestamp = System.currentTimeMillis();
+
+ String timestampStr = FORMAT.format(new Date(timestamp));
+
JSONObject jsonObject = new JSONObject();
- jsonObject.put(DataResource.VALUE, initialValue);
- jsonObject.put("device", serviceProvider);
- jsonObject.put("service", service);
- jsonObject.put("resource", resource);
- jsonObject.put(DataResource.VALUE, initialValue);
+ for(Enumeration<String> it = ts.keys();it.hasMoreElements();) {
+ String k = it.nextElement();
+ jsonObject.put(k, ts.get(k));
+ }
+ jsonObject.put(DataResource.VALUE, value);
jsonObject.put("timestamp", timestampStr);
- LOG.debug("pushing to stack {}/{}/{}/{}...", serviceProvider, service, resource, initialValue);
+
this.storageConnection.stack.push(jsonObject);
- LOG.debug("...done");
- }
-
- @Override
- public void doHandle(SnaErrorMessageImpl message) {
+
+ LOG.debug("pushed to stack : {}/{}...", path, value);
}
- @Override
- public void doHandle(SnaResponseMessage<?, ?> message) {
- }
-
- @Override
public void stop() {
- super.stop();
- this.storageConnection.close();
+ if(this.storageConnection !=null)
+ this.storageConnection.close();
}
}
diff --git a/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageConnection.java b/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageConnection.java
index 101680d..8596d04 100644
--- a/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageConnection.java
+++ b/platform/northbound/generic-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/generic/StorageConnection.java
@@ -15,10 +15,8 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-
/**
- * HTTP Agent dedicated to storage service
+ * Processes the effective storage in the linked data store
*
* @author <a href="mailto:christophe.munilla@cea.fr">Christophe Munilla</a>
*/
@@ -27,44 +25,36 @@
private static final Logger LOG = LoggerFactory.getLogger(StorageConnection.class);
protected Mediator mediator;
- protected String login, password;
protected Stack stack;
private boolean running = false;
/**
- * Send the request described by the {@link JSONObject} passed as
- * parameter
+ * Store the JSON formated data passed as parameter
*
- * @param object the {@link JSONObject} describing the request to be sent
+ * @param object the {@link JSONObject} wrapping the data to be stored
*/
- protected abstract void sendRequest(JSONObject object);
+ protected abstract void store(JSONObject object);
/**
* Constructor
*
- * @param mediator the associated {@link Mediator}
- * @param uri the string URI of the storage server
- * @param login the user login
- * @param password the user password
- * @throws IOException Exception on connection problem
+ * @param mediator the {@link Mediator} allowing the StorageConnection to be
+ * instantiated to interact with the OSGi host environment
*/
- public StorageConnection(Mediator mediator, String login, String password) throws IOException {
+ public StorageConnection(Mediator mediator) {
this.mediator = mediator;
- this.login = login;
- this.password = password;
this.stack = new Stack();
this.running = true;
Runnable runner = new Runnable() {
@Override
public void run() {
- LOG.info("POP thread started");
while (running) {
try {
JSONObject element = stack.pop();
if (element != null) {
- sendRequest(element);
+ store(element);
} else {
if (!shortSleep(200)) {
running = false;
@@ -74,7 +64,6 @@
LOG.error("POP thread error", e);
}
}
- LOG.info("POP thread terminated");
}
};
new Thread(runner).start();
@@ -82,12 +71,11 @@
protected void close() {
for (int i = 0; i < 10_000 && !this.stack.isEmpty(); i++) {
- if (!shortSleep(200)) {
- return;
- }
+ if (!shortSleep(200))
+ break;
}
- LOG.info("close operation ended");
this.running = false;
+ LOG.info("close operation ended");
}
private boolean shortSleep(long millis) {
@@ -95,7 +83,6 @@
Thread.sleep(millis);
return true;
} catch (InterruptedException e) {
- LOG.error("Sleep operation error", e);
Thread.interrupted();
return false;
}
diff --git a/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/HttpStorageAgent.java b/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/HttpStorageAgent.java
new file mode 100644
index 0000000..ab18a86
--- /dev/null
+++ b/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/HttpStorageAgent.java
@@ -0,0 +1,53 @@
+package org.eclipse.sensinact.gateway.agent.storage.http;
+
+import java.io.IOException;
+import java.util.regex.Pattern;
+
+import org.eclipse.sensinact.gateway.agent.storage.generic.StorageAgent;
+import org.eclipse.sensinact.gateway.agent.storage.http.internal.HttpStorageConnection;
+import org.eclipse.sensinact.gateway.common.bundle.Mediator;
+import org.eclipse.sensinact.gateway.core.message.AgentRelay;
+import org.osgi.framework.BundleContext;
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+
+@Component(immediate=true, service = {AgentRelay.class})
+public class HttpStorageAgent extends StorageAgent {
+
+ private static final Pattern PATTERN = Pattern.compile("^http[s]*://.*/write/measure$");
+
+ private String login;
+ private String password;
+ private String broker;
+
+ private Mediator mediator;
+
+ @Activate
+ public void activate(ComponentContext context) {
+ BundleContext bc = context.getBundleContext();
+ this.mediator = new Mediator(bc);
+
+ this.login = (String) mediator.getProperty("login");
+ this.password = (String) mediator.getProperty("password");
+ this.broker = (String) mediator.getProperty("broker");
+
+ if(!PATTERN.matcher(this.broker).matches()) {
+ context.getComponentInstance().dispose();
+ return;
+ }
+
+ try {
+ super.setStorageConnection(new HttpStorageConnection(this.mediator, broker, login, password));
+ } catch (IOException e) {
+ mediator.error(e);
+ context.getComponentInstance().dispose();
+ }
+ }
+
+ @Deactivate
+ public void deactivate() {
+ super.stop();
+ }
+}
diff --git a/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/internal/HttpStorageConnection.java b/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/internal/HttpStorageConnection.java
index 3d77191..3fda572 100644
--- a/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/internal/HttpStorageConnection.java
+++ b/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/internal/HttpStorageConnection.java
@@ -37,9 +37,12 @@
public class HttpStorageConnection extends StorageConnection {
private static final Logger LOG = LoggerFactory.getLogger(HttpStorageConnection.class);
- protected String uri;
+ protected String broker;
protected String authorization;
+ private String login;
+ private String password;
+
/**
* Constructor
*
@@ -50,21 +53,23 @@
* @throws IOException Exception on connection problem
*/
public HttpStorageConnection(Mediator mediator, String uri, String login, String password) throws IOException {
- super(mediator, login, password);
- this.uri = uri;
- this.authorization = Base64.encodeBytes((super.login + ":" + super.password).getBytes());
+ super(mediator);
+ this.login = login;
+ this.password = password;
+ this.broker = uri;
+ this.authorization = Base64.encodeBytes((this.login + ":" + this.password).getBytes());
}
/**
* Executes the HTTP request defined by the method, target, headers and entity
* arguments
*/
- protected void sendRequest(JSONObject object) {
+ protected void store(JSONObject object) {
ConnectionConfiguration<SimpleResponse, SimpleRequest> configuration = new ConnectionConfigurationImpl<>();
try {
configuration.setContentType("application/json");
configuration.setAccept("application/json");
- configuration.setUri(this.uri);
+ configuration.setUri(this.broker);
configuration.setContent(object.toString());
configuration.setHttpMethod("POST");
configuration.addHeader("Authorization", "Basic " + authorization);
diff --git a/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/osgi/Activator.java b/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/osgi/Activator.java
deleted file mode 100644
index 420be3f..0000000
--- a/platform/northbound/http-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/http/osgi/Activator.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
-* Copyright (c) 2020 Kentyou.
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * which accompanies this distribution, and is available at
- * http://www.eclipse.org/legal/epl-v10.html
- *
- * Contributors:
-* Kentyou - initial API and implementation
- */
-package org.eclipse.sensinact.gateway.agent.storage.http.osgi;
-
-import org.eclipse.sensinact.gateway.agent.storage.http.internal.HttpStorageConnection;
-import org.eclipse.sensinact.gateway.agent.storage.generic.StorageAgent;
-import org.eclipse.sensinact.gateway.common.annotation.Property;
-import org.eclipse.sensinact.gateway.common.bundle.AbstractActivator;
-import org.eclipse.sensinact.gateway.common.bundle.Mediator;
-import org.eclipse.sensinact.gateway.common.execution.Executable;
-import org.eclipse.sensinact.gateway.core.Core;
-import org.osgi.framework.BundleContext;
-
-/**
- * Extended {@link AbstractActivator}
- */
-public class Activator extends AbstractActivator<Mediator> {
- @Property
- private String login;
- @Property
- private String password;
- @Property(validationRegex = "^http[s]*://.*/write/measure$")
- private String broker;
- private StorageAgent handler;
- private String registration;
-
- @Override
- public void doStart() throws Exception {
- try {
- if (super.mediator.isDebugLoggable())
- super.mediator.debug("Starting storage agent.");
- this.handler = new StorageAgent(new HttpStorageConnection(super.mediator, broker, login, password));
- this.registration = mediator.callService(Core.class, new Executable<Core, String>() {
- @Override
- public String execute(Core service) throws Exception {
- return service.registerAgent(mediator, Activator.this.handler, null);
- }
- });
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- @Override
- public void doStop() throws Exception {
- if (super.mediator.isDebugLoggable()) {
- super.mediator.debug("Stopping storage agent.");
- }
- if(this.handler!=null){
- this.handler.stop();
- }
- this.registration = null;
- this.handler = null;
- }
-
- @Override
- public Mediator doInstantiate(BundleContext context) {
- return new Mediator(context);
- }
-}
diff --git a/platform/northbound/influxdb-storage-agent/pom.xml b/platform/northbound/influxdb-storage-agent/pom.xml
index 3c1314f..91836f2 100644
--- a/platform/northbound/influxdb-storage-agent/pom.xml
+++ b/platform/northbound/influxdb-storage-agent/pom.xml
@@ -31,6 +31,11 @@
<scope>provided</scope>
</dependency>
<dependency>
+ <groupId>org.eclipse.sensinact.gateway.tools</groupId>
+ <artifactId>influxdb-connector</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
<groupId>org.eclipse.sensinact.gateway</groupId>
<artifactId>sensinact-common</artifactId>
<scope>provided</scope>
@@ -45,13 +50,6 @@
<artifactId>sensinact-core</artifactId>
<scope>provided</scope>
</dependency>
- <!-- https://mvnrepository.com/artifact/org.apache.servicemix.bundles/org.apache.servicemix.bundles.influxdb-java -->
- <dependency>
- <groupId>org.influxdb</groupId>
- <artifactId>influxdb-java</artifactId>
- <version>2.15</version>
- <type>jar</type>
- </dependency>
</dependencies>
<build>
@@ -60,143 +58,8 @@
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
<extensions>true</extensions>
- <configuration>
- <instructions>
- <Bundle-Activator>org.eclipse.sensinact.gateway.southbound.influxdb.osgi.Activator</Bundle-Activator>
- <Private-Package>
- org.influxdb,
- org.influxdb.*
- </Private-Package>
- </instructions>
- </configuration>
</plugin>
</plugins>
</build>
-
- <profiles>
- <profile>
- <id>build-for-felix</id>
- <dependencies>
- <dependency>
- <groupId>org.apache.felix</groupId>
- <artifactId>org.apache.felix.main</artifactId>
- <version>4.0.3</version>
- <scope>provided</scope>
- </dependency>
- <!-- To include a shell:
- <dependency>
- <groupId>org.apache.felix</groupId>
- <artifactId>org.apache.felix.gogo.shell</artifactId>
- <version>0.10.0</version>
- </dependency>
- -->
- </dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <version>1.7</version>
- <executions>
- <execution>
- <id>compile</id>
- <phase>package</phase>
- <goals>
- <goal>run</goal>
- </goals>
- <configuration>
- <target>
- <pathconvert property="plugins.jars" pathsep="${path.separator}">
- <path refid="maven.runtime.classpath"/>
- <map from="${project.build.directory}${file.separator}classes" to=""/>
- </pathconvert>
- <pathconvert pathsep=" " property="bundles">
- <path path="${plugins.jars}"/>
- <mapper>
- <chainedmapper>
- <flattenmapper/>
- <globmapper from="*" to="file:modules/*" casesensitive="no"/>
- </chainedmapper>
- </mapper>
- </pathconvert>
- <propertyfile file="${project.build.directory}/config.properties">
- <entry key="felix.auto.start" value="${bundles} file:modules/${project.build.finalName}.jar"/>
- <entry key="org.osgi.framework.bootdelegation" value="*"/>
- </propertyfile>
- <copy file="${maven.dependency.org.apache.felix.org.apache.felix.main.jar.path}" tofile="${project.build.directory}/felix.jar"/>
- </target>
-
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-assembly-plugin</artifactId>
- <version>2.3</version>
- <executions>
- <execution>
- <id>create-executable-jar</id>
- <phase>package</phase>
- <goals>
- <goal>single</goal>
- </goals>
- <configuration>
- <descriptors>
- <descriptor>${basedir}/src/main/assembly/felix.xml</descriptor>
- </descriptors>
- <finalName>${project.build.finalName}</finalName>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
- </profile>
- <profile>
- <id>run-on-felix</id>
- <dependencies>
- <dependency>
- <groupId>org.apache.felix</groupId>
- <artifactId>org.apache.felix.main</artifactId>
- <version>4.0.3</version>
- <scope>provided</scope>
- </dependency>
- <!-- org.apache.felix:org.apache.felix.gogo.shell:0.6.1 useless from Maven since stdin is swallowed -->
- </dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <version>1.7</version>
- <configuration>
- <target>
- <property name="vm.args" value=""/>
- <pathconvert property="plugins.jars" pathsep="${path.separator}">
- <path refid="maven.runtime.classpath"/>
- <map from="${project.build.directory}${file.separator}classes" to=""/>
- </pathconvert>
- <makeurl property="urls" separator=" ">
- <path path="${plugins.jars}"/>
- <path location="${project.build.directory}/${project.build.finalName}.jar"/>
- </makeurl>
- <propertyfile file="${project.build.directory}/run.properties">
- <entry key="felix.auto.start" value="${urls}"/>
- <entry key="felix.auto.deploy.action" value="uninstall,install,update,start"/>
- <entry key="org.osgi.framework.storage" value="${project.build.directory}${file.separator}felix-cache"/>
- <entry key="org.osgi.framework.bootdelegation" value="*"/>
- </propertyfile>
- <makeurl property="run.properties.url" file="${project.build.directory}/run.properties"/>
- <java fork="true" jar="${maven.dependency.org.apache.felix.org.apache.felix.main.jar.path}">
- <sysproperty key="felix.config.properties" value="${run.properties.url}"/>
- <jvmarg line="${vm.args}"/>
- </java>
- </target>
- </configuration>
- </plugin>
- </plugins>
- </build>
- </profile>
- </profiles>
+
</project>
diff --git a/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/InfluxDBStorageAgent.java b/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/InfluxDBStorageAgent.java
new file mode 100644
index 0000000..2e0a323
--- /dev/null
+++ b/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/InfluxDBStorageAgent.java
@@ -0,0 +1,91 @@
+package org.eclipse.sensinact.gateway.agent.storage.influxdb;
+
+import java.io.IOException;
+
+import org.eclipse.sensinact.gateway.agent.storage.generic.StorageAgent;
+import org.eclipse.sensinact.gateway.agent.storage.influxdb.internal.InfluxDBStorageConnection;
+import org.eclipse.sensinact.gateway.common.bundle.Mediator;
+import org.eclipse.sensinact.gateway.core.message.AgentRelay;
+import org.eclipse.sensinact.gateway.tools.connector.influxdb.InfluxDbConnector;
+import org.eclipse.sensinact.gateway.tools.connector.influxdb.InfluxDbConnectorConfiguration;
+import org.eclipse.sensinact.gateway.tools.connector.influxdb.InfluxDbDatabase;
+import org.osgi.framework.BundleContext;
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+
+@Component(immediate=true, service = {AgentRelay.class})
+public class InfluxDBStorageAgent extends StorageAgent {
+
+ private static final String INFLUX_AGENT_SCHEME_PROPS = "org.eclipse.sensinact.gateway.history.influx.scheme";
+ private static final String INFLUX_AGENT_HOST_PROPS = "org.eclipse.sensinact.gateway.history.influx.host";
+ private static final String INFLUX_AGENT_PORT_PROPS = "org.eclipse.sensinact.gateway.history.influx.port";
+ private static final String INFLUX_AGENT_PATH_PROPS = "org.eclipse.sensinact.gateway.history.influx.path";
+
+ private static final String INFLUX_AGENT_LOGIN_PROPS = "org.eclipse.sensinact.gateway.history.influx.login";
+ private static final String INFLUX_AGENT_PASSWORD_PROPS = "org.eclipse.sensinact.gateway.history.influx.password";
+
+
+ private static final String INFLUX_AGENT_MEASUREMENT_PROPS = "org.eclipse.sensinact.gateway.history.influx.measurement";
+ private static final String INFLUX_AGENT_DEFAULT_MEASUREMENT = "test";
+
+ private Mediator mediator;
+ private InfluxDbConnector connector;
+ private String measurement;
+ private InfluxDbDatabase database;
+
+ @Activate
+ public void activate(ComponentContext context) {
+ BundleContext bc = context.getBundleContext();
+ this.mediator = new Mediator(bc);
+
+ String scheme = (String) mediator.getProperty(INFLUX_AGENT_SCHEME_PROPS);
+ if(scheme == null)
+ scheme = InfluxDbConnectorConfiguration.DEFAULT_SCHEME;
+
+ String host = (String) mediator.getProperty(INFLUX_AGENT_HOST_PROPS);
+ if(host == null)
+ host = InfluxDbConnectorConfiguration.DEFAULT_HOST;
+
+ int port = -1;
+ String portStr = (String) mediator.getProperty(INFLUX_AGENT_PORT_PROPS);
+ if(portStr == null)
+ port = InfluxDbConnectorConfiguration.DEFAULT_PORT;
+ else
+ port = Integer.parseInt(portStr);
+
+ String path = (String) mediator.getProperty(INFLUX_AGENT_PATH_PROPS);
+ if(path == null)
+ path = InfluxDbConnectorConfiguration.DEFAULT_PATH;
+
+ String username = (String) mediator.getProperty(INFLUX_AGENT_LOGIN_PROPS);
+ String password = (String) mediator.getProperty(INFLUX_AGENT_PASSWORD_PROPS);
+
+ InfluxDbConnectorConfiguration configuration = new InfluxDbConnectorConfiguration.Builder(
+ ).withScheme(scheme
+ ).withHost(host
+ ).withPort(port
+ ).withPath(path
+ ).withUsername(username
+ ).withPassword(password
+ ).build();
+ try {
+ this.connector = new InfluxDbConnector(configuration);
+ } catch (IOException e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ this.measurement = (String) mediator.getProperty(INFLUX_AGENT_MEASUREMENT_PROPS);
+ if(this.measurement == null)
+ this.measurement = INFLUX_AGENT_DEFAULT_MEASUREMENT;
+
+ this.database = this.connector.createIfNotExists("sensinact");
+ super.setStorageConnection(new InfluxDBStorageConnection(mediator, database, this.measurement));
+ }
+
+ @Deactivate
+ public void deactivate() {
+ super.stop();
+ }
+}
diff --git a/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/internal/InfluxDBStorageConnection.java b/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/internal/InfluxDBStorageConnection.java
new file mode 100644
index 0000000..a4bb1e3
--- /dev/null
+++ b/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/internal/InfluxDBStorageConnection.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright (c) 2020 Kentyou.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ * Kentyou - initial API and implementation
+ */
+package org.eclipse.sensinact.gateway.agent.storage.influxdb.internal;
+
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
+
+import org.eclipse.sensinact.gateway.agent.storage.generic.StorageConnection;
+import org.eclipse.sensinact.gateway.common.bundle.Mediator;
+import org.eclipse.sensinact.gateway.core.DataResource;
+import org.eclipse.sensinact.gateway.tools.connector.influxdb.InfluxDbDatabase;
+import org.eclipse.sensinact.gateway.util.json.JSONObjectStatement;
+import org.eclipse.sensinact.gateway.util.json.JSONTokenerStatement;
+import org.json.JSONObject;
+
+/**
+ * Extended {@link SorageConnection} dedicated to InfluxDB data store
+ */
+public class InfluxDBStorageConnection extends StorageConnection {
+
+ private static final JSONObjectStatement STATEMENT =
+ new JSONObjectStatement(new JSONTokenerStatement(
+ "{" +
+ " \"type\": \"Feature\"," +
+ " \"properties\": {" +
+ " \"name\": $(name)" +
+ " }," +
+ " \"geometry\": {" +
+ " \"type\": \"Point\"," +
+ " \"coordinates\": [ $(longitude),$(latitude)] " +
+ " }" +
+ "}"));
+
+ private String measurement;
+ private InfluxDbDatabase database;
+
+ /**
+ * Constructor
+ *
+ * @param mediator the {@link Mediator} allowing the InfluxDbAgentCallback to be instantiated
+ * to interact with the OSGi host environment
+ * @param database the {@link InfluxDbDatabase} in which data will be stored
+ * @param measurement the String name of the measurement in which data will be stored
+ *
+ * @throws IOException
+ */
+ public InfluxDBStorageConnection(Mediator mediator, InfluxDbDatabase database, String measurement){
+ super(mediator);
+ this.database = database;
+ this.measurement = measurement;
+ }
+
+ @Override
+ public void store(JSONObject obj) {
+ String uri = (String) obj.opt("path");
+ String location = (String) obj.opt("location");
+ double latitude = -1;
+ double longitude = -1;
+ String geolocation = null;
+
+ if(location != null) {
+ String[] locationElements = location.split(":");
+ if(locationElements.length == 2) {
+ latitude = Double.parseDouble(locationElements[0]);
+ longitude = Double.parseDouble(locationElements[1]);
+ STATEMENT.apply("latitude", latitude);
+ STATEMENT.apply("longitude", longitude);
+ STATEMENT.apply("name", obj.opt("provider"));
+ geolocation = STATEMENT.toString();
+ } else
+ geolocation = location;
+ }
+ final Dictionary<String,String> ts = new Hashtable<>();
+ ts.put("path",uri);
+ ts.put("latitude", latitude==-1?"N/A":String.valueOf(latitude));
+ ts.put("longitude", longitude==-1?"N/A":String.valueOf(longitude));
+ ts.put("geolocation", geolocation==null?"N/A":geolocation);
+
+ this.database.add(measurement, ts, obj.opt(DataResource.VALUE));
+ }
+}
diff --git a/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/internal/InfluxStorageConnection.java b/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/internal/InfluxStorageConnection.java
deleted file mode 100644
index c0831c7..0000000
--- a/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/internal/InfluxStorageConnection.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.eclipse.sensinact.gateway.agent.storage.influxdb.internal;
-
-import java.io.IOException;
-import java.util.concurrent.TimeUnit;
-import org.eclipse.sensinact.gateway.common.bundle.Mediator;
-import org.eclipse.sensinact.gateway.agent.storage.generic.StorageConnection;
-import org.influxdb.InfluxDB;
-import org.influxdb.InfluxDBFactory;
-import org.influxdb.dto.BatchPoints;
-import org.influxdb.dto.Point;
-import org.influxdb.dto.Point.Builder;
-import org.influxdb.dto.Pong;
-import org.json.JSONObject;
-
-/**
- *
- * @author kleisar
- */
-public class InfluxStorageConnection extends StorageConnection {
-
- private InfluxDB influxDB;
- private BatchPoints batchPoints;
-
- protected String databaseURL;
-
- public InfluxStorageConnection(Mediator mediator, String databaseURL, String userName, String password) throws IOException {
- super(mediator, userName, password);
- this.databaseURL = databaseURL;
- this.connect();
- }
-
- private void insert(String provider, String service, String resource, Object value) {
- System.out.println("insert " + provider + ", " + service + ", " + resource + " = " + value);
- Point point = null;
- Builder builder = Point.measurement("sensiNact")
- .tag("provider", provider)
- .tag("service", service)
- .tag("resource", resource);
- if (value instanceof String) {
- point = builder.addField("value", (String) value).build();
- } else if (value instanceof Double) {
- point = builder.addField("value", (Double) value).build();
- } else if (value instanceof Float) {
- point = builder.addField("value", (Float) value).build();
- } else if (value instanceof Integer) {
- point = builder.addField("value", (Integer) value).build();
- }
-
- influxDB.write(point);
- //batchPoints.point(point);
- }
-
- @Override
- public void sendRequest(JSONObject object) {
- try {
- String device = object.getString("device");
- String service = object.getString("service");
- String resource = object.getString("resource");
- Object value = object.get("value");
- this.insert(device, service, resource, value);
- } catch (Exception ex) {
- ex.printStackTrace();
- }
- }
-
- /**
- * Creates a connection and returns true if it has been done properly.
- * Returns false otherwise
- *
- * @return <ul>
- * <li>true if the connection has been properly done</li>
- * <li>false otherwise</li>
- * </ul>
- */
- protected boolean connect() {
- influxDB = InfluxDBFactory.connect(this.databaseURL);
- influxDB.setDatabase("sensiNact");
- influxDB.setRetentionPolicy("autogen");
- Pong response = this.influxDB.ping();
- if (response.getVersion().equalsIgnoreCase("unknown")) {
- System.out.println("Error pinging server.");
- return false;
- } else {
-// this.influxDB.query(new Query("CREATE DATABASE sensiNact", ""));
-// batchPoints = BatchPoints
-// .database("sensiNact")
-// .retentionPolicy("defaultPolicy")
-// .build();
- }
- influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
- influxDB.enableBatch(100, 5000, TimeUnit.MILLISECONDS); //wait 5s or 100Points, whichever occurs first before sending the points to influxdb
- return true;
- }
-
-}
diff --git a/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/osgi/Activator.java b/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/osgi/Activator.java
deleted file mode 100644
index c4d613c..0000000
--- a/platform/northbound/influxdb-storage-agent/src/main/java/org/eclipse/sensinact/gateway/agent/storage/influxdb/osgi/Activator.java
+++ /dev/null
@@ -1,67 +0,0 @@
-package org.eclipse.sensinact.gateway.agent.storage.influxdb.osgi;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.Properties;
-
-import org.eclipse.sensinact.gateway.common.bundle.AbstractActivator;
-import org.eclipse.sensinact.gateway.common.bundle.Mediator;
-import org.eclipse.sensinact.gateway.common.execution.Executable;
-import org.eclipse.sensinact.gateway.core.Core;
-import org.eclipse.sensinact.gateway.agent.storage.generic.StorageAgent;
-import org.eclipse.sensinact.gateway.agent.storage.influxdb.internal.InfluxStorageConnection;
-import org.osgi.framework.BundleContext;
-
-public class Activator extends AbstractActivator<Mediator> {
-
- private String databaseURL;
- private String login;
- private String password;
- private StorageAgent handler;
- private String registration;
-
- @Override
- public void doStart() throws Exception {
- try {
- if (super.mediator.isDebugLoggable()) {
- super.mediator.debug("Starting storage agent.");
- }
- Properties prop = new Properties();
- try {
- prop.load(new FileInputStream("cfgs/influxdb.property"));
- //sslClientProperties.load(new FileInputStream("cfgs/sslproperties.property"));
- login = prop.getProperty("username");
- password = prop.getProperty("password");
- databaseURL = prop.getProperty("databaseURL");
- } catch (IOException ex) {
- mediator.error(ex);
- }
- this.handler = new StorageAgent(new InfluxStorageConnection(super.mediator, databaseURL, login, password));
- this.registration = mediator.callService(Core.class, new Executable<Core, String>() {
- @Override
- public String execute(Core service) throws Exception {
- return service.registerAgent(mediator, Activator.this.handler, null);
- }
- });
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- @Override
- public void doStop() throws Exception {
- if (super.mediator.isDebugLoggable()) {
- super.mediator.debug("Stopping storage agent.");
- }
- if (this.handler != null) {
- this.handler.stop();
- }
- this.registration = null;
- this.handler = null;
- }
-
- @Override
- public Mediator doInstantiate(BundleContext context) {
- return new Mediator(context);
- }
-}
