added analytics extension to persist event streams,receivers,publishers

master
Amalka Subasinghe 2 years ago
parent ad3cc84558
commit 766606f9f7

@ -0,0 +1,119 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
~
~ Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
~ Version 2.0 (the "License"); you may not use this file except
~ in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing,
~ software distributed under the License is distributed on an
~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
~ KIND, either express or implied. See the License for the
~ specific language governing permissions and limitations
~ under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>apimgt-extensions</artifactId>
<groupId>org.wso2.carbon.devicemgt</groupId>
<version>5.0.21-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>io.entgra.device.mgt.core.apimgt.analytics.extension</artifactId>
<packaging>bundle</packaging>
<name>Entgra - API mgt analytics extension</name>
<url>http://wso2.org</url>
<dependencies>
<dependency>
<groupId>org.wso2.carbon</groupId>
<artifactId>org.wso2.carbon.core</artifactId>
</dependency>
<dependency>
<groupId>org.wso2.carbon</groupId>
<artifactId>org.wso2.carbon.utils</artifactId>
</dependency>
<dependency>
<groupId>org.apache.velocity</groupId>
<artifactId>velocity</artifactId>
<version>1.7</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<excludes>
<exclude>**/Abstract*</exclude>
</excludes>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-scr-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
<extensions>true</extensions>
<configuration>
<instructions>
<Bundle-SymbolicName>${project.artifactId}</Bundle-SymbolicName>
<Bundle-Name>${project.artifactId}</Bundle-Name>
<Bundle-Version>${carbon.device.mgt.version}</Bundle-Version>
<Bundle-Description>API Management Application Bundle</Bundle-Description>
<Private-Package>org.wso2.carbon.apimgt.application.extension.internal</Private-Package>
<Import-Packages>
io.entgra.device.mgt.core.apimgt.analytics.extension.dto,
org.apache.velocity,
org.apache.velocity.app,
org.apache.velocity.context,
org.wso2.carbon.utils;version="[4.6,5)"
</Import-Packages>
<Export-Package>
io.entgra.device.mgt.core.apimgt.analytics.extension.*
</Export-Package>
<Embed-Dependency>
scribe;scope=compile|runtime;inline=false;
</Embed-Dependency>
</instructions>
</configuration>
</plugin>
<plugin>
<groupId>org.jacoco</groupId>
<artifactId>jacoco-maven-plugin</artifactId>
<configuration>
<destFile>${basedir}/target/coverage-reports/jacoco-unit.exec</destFile>
</configuration>
<executions>
<execution>
<id>jacoco-initialize</id>
<goals>
<goal>prepare-agent</goal>
</goals>
</execution>
<execution>
<id>jacoco-site</id>
<phase>test</phase>
<goals>
<goal>report</goal>
</goals>
<configuration>
<dataFile>${basedir}/target/coverage-reports/jacoco-unit.exec</dataFile>
<outputDirectory>${basedir}/target/coverage-reports/site</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

@ -0,0 +1,169 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension;
import io.entgra.device.mgt.core.apimgt.analytics.extension.dto.*;
import io.entgra.device.mgt.core.apimgt.analytics.extension.exception.EventPublisherDeployerException;
import io.entgra.device.mgt.core.apimgt.analytics.extension.exception.EventReceiverDeployerException;
import io.entgra.device.mgt.core.apimgt.analytics.extension.exception.EventStreamDeployerException;
import org.apache.velocity.Template;
import org.apache.velocity.VelocityContext;
import org.apache.velocity.app.VelocityEngine;
import org.apache.velocity.runtime.RuntimeConstants;
import org.wso2.carbon.utils.CarbonUtils;
import org.wso2.carbon.utils.multitenancy.MultitenantConstants;
import java.io.*;
public class AnalyticsArtifactsDeployer {
public static final String TEMPLATE_LOCATION = "repository" + File.separator + "resources" + File.separator + "io.entgra.device.mgt.core.apimgt.analytics.extension/src/main/resources/iot-analytics-templates";
public static final String EVENT_STREAM_LOCATION = "eventstreams";
public static final String EVENT_PUBLISHER_LOCATION = "eventpublishers";
public static final String EVENT_RECEIVER_LOCATION = "eventreceivers";
public static final String EVENT_STREAM_TEMPLATE = TEMPLATE_LOCATION + File.separator + "event_stream.json.template";
public static final String EVENT_PUBLISHER_TEMPLATE = TEMPLATE_LOCATION + File.separator + "event_publisher.xml.template";
public static final String EVENT_RECEIVER_TEMPLATE = TEMPLATE_LOCATION + File.separator + "event_receiver.xml.template";
public void deployEventStream(EventStreamData eventStreamData, int tenantId) throws EventStreamDeployerException {
try {
VelocityEngine ve = new VelocityEngine();
ve.setProperty(RuntimeConstants.FILE_RESOURCE_LOADER_PATH, CarbonUtils.getCarbonHome());
ve.init();
Template template = ve.getTemplate(EVENT_STREAM_TEMPLATE);
VelocityContext context = populateContextForEventStreams(eventStreamData);
StringWriter writer = new StringWriter();
template.merge(context, writer);
String fileName = eventStreamData.getName() + "_" + eventStreamData.getVersion() + ".json";
String fileLocation = null;
if (MultitenantConstants.SUPER_TENANT_ID == tenantId) {
fileLocation = CarbonUtils.getCarbonHome() + File.separator + "repository" + File.separator + "deployment"
+ File.separator + "server" + File.separator + EVENT_STREAM_LOCATION + File.separator + fileName;
} else {
fileLocation = CarbonUtils.getCarbonTenantsDirPath() + File.separator + tenantId + File.separator
+ EVENT_STREAM_LOCATION + File.separator + fileName;
}
PrintWriter printWriter = new PrintWriter(fileLocation, "UTF-8");
printWriter.println(writer.toString());
printWriter.close();
} catch (FileNotFoundException | UnsupportedEncodingException e) {
throw new EventStreamDeployerException("Error while persisting event stream definition ", e);
}
}
public void deployEventPublisher(EventPublisherData eventPublisherData, int tenantId) throws EventPublisherDeployerException {
try {
VelocityEngine ve = new VelocityEngine();
ve.setProperty(RuntimeConstants.FILE_RESOURCE_LOADER_PATH, CarbonUtils.getCarbonHome());
ve.init();
Template template = ve.getTemplate(EVENT_PUBLISHER_TEMPLATE);
VelocityContext context = populateContextForEventPublisher(eventPublisherData);
StringWriter writer = new StringWriter();
template.merge(context, writer);
String fileName = eventPublisherData.getName() + ".xml";
String fileLocation = null;
if (MultitenantConstants.SUPER_TENANT_ID == tenantId) {
fileLocation = CarbonUtils.getCarbonHome() + File.separator + "repository" + File.separator + "deployment"
+ File.separator + "server" + File.separator + EVENT_PUBLISHER_LOCATION + File.separator + fileName;
} else {
fileLocation = CarbonUtils.getCarbonTenantsDirPath() + File.separator + tenantId + File.separator
+ EVENT_PUBLISHER_LOCATION + File.separator + fileName;
}
PrintWriter printWriter = new PrintWriter(fileLocation, "UTF-8");
printWriter.println(writer.toString());
printWriter.close();
} catch (FileNotFoundException | UnsupportedEncodingException e) {
throw new EventPublisherDeployerException("Error while persisting rdbms event publisher ", e);
}
}
public void deployEventReceiver(EventReceiverData eventReceiverData, int tenantId) throws EventReceiverDeployerException {
try {
VelocityEngine ve = new VelocityEngine();
ve.setProperty(RuntimeConstants.FILE_RESOURCE_LOADER_PATH, CarbonUtils.getCarbonHome());
ve.init();
Template template = ve.getTemplate(EVENT_RECEIVER_TEMPLATE);
VelocityContext context = populateContextForEventReceiver(eventReceiverData);
StringWriter writer = new StringWriter();
template.merge(context, writer);
String fileName = eventReceiverData.getName() + ".xml";
String fileLocation = null;
if (MultitenantConstants.SUPER_TENANT_ID == tenantId) {
fileLocation = CarbonUtils.getCarbonHome() + File.separator + "repository" + File.separator + "deployment"
+ File.separator + "server" + File.separator + EVENT_RECEIVER_LOCATION + File.separator + fileName;
} else {
fileLocation = CarbonUtils.getCarbonTenantsDirPath() + File.separator + tenantId + File.separator
+ EVENT_RECEIVER_LOCATION + File.separator + fileName;
}
PrintWriter printWriter = new PrintWriter(fileLocation, "UTF-8");
printWriter.println(writer.toString());
printWriter.close();
} catch (FileNotFoundException | UnsupportedEncodingException e) {
throw new EventReceiverDeployerException("Error while persisting oauth mqtt event receiver ", e);
}
}
private VelocityContext populateContextForEventStreams(EventStreamData eventStreamData) {
VelocityContext context = new VelocityContext();
context.put("name", eventStreamData.getName());
context.put("version", eventStreamData.getVersion());
context.put("metaData",
eventStreamData.getMetaData() != null ? eventStreamData.getMetaData() : new MetaData("deviceId", "STRING"));
if (eventStreamData.getPayloadData() != null) {
context.put("properties", eventStreamData.getPayloadData());
}
return context;
}
private VelocityContext populateContextForEventPublisher(EventPublisherData eventPublisherData) {
VelocityContext context = new VelocityContext();
context.put("name", eventPublisherData.getName());
context.put("streamName", eventPublisherData.getStreamName());
context.put("streamVersion", eventPublisherData.getStreamVersion());
context.put("properties", eventPublisherData.getPropertyList());
context.put("eventAdapterType", eventPublisherData.getEventAdaptorType());
context.put("customMappingType", eventPublisherData.getCustomMappingType());
return context;
}
private VelocityContext populateContextForEventReceiver(EventReceiverData eventReceiverData) {
VelocityContext context = new VelocityContext();
context.put("name", eventReceiverData.getName());
context.put("streamName", eventReceiverData.getStreamName());
context.put("streamVersion", eventReceiverData.getStreamVersion());
context.put("properties", eventReceiverData.getPropertyList());
context.put("eventAdapterType", eventReceiverData.getEventAdapterType());
context.put("customMappingType", eventReceiverData.getCustomMappingType());
return context;
}
}

@ -0,0 +1,80 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.dto;
import java.util.List;
public class EventPublisherData {
private String name;
private String streamVersion;
private String streamName;
private List<Property> propertyList;
private String eventAdaptorType;
private String customMappingType;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getStreamVersion() {
return streamVersion;
}
public void setStreamVersion(String streamVersion) {
this.streamVersion = streamVersion;
}
public String getStreamName() {
return streamName;
}
public void setStreamName(String streamName) {
this.streamName = streamName;
}
public List<Property> getPropertyList() {
return propertyList;
}
public void setPropertyList(List<Property> propertyList) {
this.propertyList = propertyList;
}
public String getEventAdaptorType() {
return eventAdaptorType;
}
public void setEventAdaptorType(String eventAdaptorType) {
this.eventAdaptorType = eventAdaptorType;
}
public String getCustomMappingType() {
return customMappingType;
}
public void setCustomMappingType(String customMappingType) {
this.customMappingType = customMappingType;
}
}

@ -0,0 +1,81 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.dto;
import java.util.List;
public class EventReceiverData {
private String name;
private String streamVersion;
private String streamName;
private String eventAdapterType;
List<Property> propertyList;
private String customMappingType;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getStreamVersion() {
return streamVersion;
}
public void setStreamVersion(String streamVersion) {
this.streamVersion = streamVersion;
}
public String getStreamName() {
return streamName;
}
public void setStreamName(String streamName) {
this.streamName = streamName;
}
public String getEventAdapterType() {
return eventAdapterType;
}
public void setEventAdapterType(String eventAdapterType) {
this.eventAdapterType = eventAdapterType;
}
public List<Property> getPropertyList() {
return propertyList;
}
public void setPropertyList(List<Property> propertyList) {
this.propertyList = propertyList;
}
public String getCustomMappingType() {
return customMappingType;
}
public void setCustomMappingType(String customMappingType) {
this.customMappingType = customMappingType;
}
}

@ -0,0 +1,59 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.dto;
import java.util.List;
public class EventStreamData {
private String name;
private String version;
private MetaData metaData;
private List<Property> payloadData;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getVersion() {
return version;
}
public void setVersion(String version) {
this.version = version;
}
public MetaData getMetaData() {
return metaData;
}
public void setMetaData(MetaData metaData) {
this.metaData = metaData;
}
public List<Property> getPayloadData() {
return payloadData;
}
public void setPayloadData(List<Property> payloadData) {
this.payloadData = payloadData;
}
}

@ -0,0 +1,44 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.dto;
public class MetaData {
String name;
String type;
public MetaData(String name, String type){
this.setName(name);
this.setType(type);
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getType() {
return type;
}
public void setType(String type) {
this.type = type;
}
}

@ -0,0 +1,45 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.dto;
public class Property {
String name;
String value;
public Property(String name, String value){
this.setName(name);
this.setValue(value);
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getValue() {
return value;
}
public void setValue(String value) {
this.value = value;
}
}

@ -0,0 +1,43 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.exception;
public class EventPublisherDeployerException extends Exception {
private static final long serialVersionUID = -3151279311929070299L;
public EventPublisherDeployerException(String msg, Exception nestedEx) {
super(msg, nestedEx);
}
public EventPublisherDeployerException(String message, Throwable cause) {
super(message, cause);
}
public EventPublisherDeployerException(String msg) {
super(msg);
}
public EventPublisherDeployerException() {
super();
}
public EventPublisherDeployerException(Throwable cause) {
super(cause);
}
}

@ -0,0 +1,43 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.exception;
public class EventReceiverDeployerException extends Exception {
private static final long serialVersionUID = -3151279311929070299L;
public EventReceiverDeployerException(String msg, Exception nestedEx) {
super(msg, nestedEx);
}
public EventReceiverDeployerException(String message, Throwable cause) {
super(message, cause);
}
public EventReceiverDeployerException(String msg) {
super(msg);
}
public EventReceiverDeployerException() {
super();
}
public EventReceiverDeployerException(Throwable cause) {
super(cause);
}
}

@ -0,0 +1,43 @@
/*
* Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
*
* Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package io.entgra.device.mgt.core.apimgt.analytics.extension.exception;
public class EventStreamDeployerException extends Exception {
private static final long serialVersionUID = -3151279311929070298L;
public EventStreamDeployerException(String msg, Exception nestedEx) {
super(msg, nestedEx);
}
public EventStreamDeployerException(String message, Throwable cause) {
super(message, cause);
}
public EventStreamDeployerException(String msg) {
super(msg);
}
public EventStreamDeployerException() {
super();
}
public EventStreamDeployerException(Throwable cause) {
super(cause);
}
}

@ -39,6 +39,7 @@
<module>org.wso2.carbon.apimgt.annotations</module>
<module>org.wso2.carbon.apimgt.keymgt.extension</module>
<module>org.wso2.carbon.apimgt.keymgt.extension.api</module>
<module>io.entgra.device.mgt.core.apimgt.analytics.extension</module>
</modules>
<build>

@ -450,5 +450,10 @@
<artifactId>org.wso2.carbon.event.output.adapter.rdbms</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.wso2.carbon.devicemgt</groupId>
<artifactId>io.entgra.device.mgt.core.apimgt.analytics.extension</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
</project>

@ -1,30 +1,20 @@
package org.wso2.carbon.device.mgt.jaxrs.service.impl;
import edu.emory.mathcs.backport.java.util.Arrays;
import io.entgra.device.mgt.core.apimgt.analytics.extension.AnalyticsArtifactsDeployer;
import io.entgra.device.mgt.core.apimgt.analytics.extension.dto.*;
import io.entgra.device.mgt.core.apimgt.analytics.extension.exception.EventReceiverDeployerException;
import io.entgra.device.mgt.core.apimgt.analytics.extension.exception.EventPublisherDeployerException;
import io.entgra.device.mgt.core.apimgt.analytics.extension.exception.EventStreamDeployerException;
import org.apache.axis2.AxisFault;
import org.apache.axis2.client.Stub;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.velocity.util.ArrayListWrapper;
import org.json.JSONObject;
import org.opensaml.xml.signature.J;
import org.wso2.carbon.analytics.stream.persistence.stub.EventStreamPersistenceAdminService;
import org.wso2.carbon.analytics.stream.persistence.stub.EventStreamPersistenceAdminServiceEventStreamPersistenceAdminServiceExceptionException;
import org.wso2.carbon.analytics.stream.persistence.stub.EventStreamPersistenceAdminServiceStub;
import org.wso2.carbon.analytics.stream.persistence.stub.dto.AnalyticsTable;
import org.wso2.carbon.analytics.stream.persistence.stub.dto.AnalyticsTableRecord;
import org.wso2.carbon.base.MultitenantConstants;
import org.wso2.carbon.context.PrivilegedCarbonContext;
import org.wso2.carbon.databridge.commons.StreamDefinition;
import org.wso2.carbon.databridge.commons.exception.MalformedStreamDefinitionException;
import org.wso2.carbon.device.mgt.common.exceptions.DeviceManagementException;
import org.wso2.carbon.device.mgt.core.dto.DeviceType;
import org.wso2.carbon.device.mgt.core.service.DeviceManagementProviderService;
import org.wso2.carbon.device.mgt.jaxrs.beans.analytics.Attribute;
import org.wso2.carbon.device.mgt.jaxrs.beans.analytics.AttributeType;
import org.wso2.carbon.device.mgt.jaxrs.beans.analytics.DeviceTypeEvent;
@ -34,24 +24,16 @@ import org.wso2.carbon.device.mgt.jaxrs.service.api.DeviceEventManagementService
import org.wso2.carbon.device.mgt.jaxrs.util.Constants;
import org.wso2.carbon.device.mgt.jaxrs.util.DeviceMgtAPIUtils;
import org.wso2.carbon.event.input.adapter.core.InputEventAdapterConfiguration;
import org.wso2.carbon.event.input.adapter.core.MessageType;
import org.wso2.carbon.event.output.adapter.core.OutputEventAdapterConfiguration;
import org.wso2.carbon.event.output.adapter.core.OutputEventAdapterService;
import org.wso2.carbon.event.output.adapter.rdbms.RDBMSEventAdapter;
import org.wso2.carbon.event.output.adapter.rdbms.internal.ds.RDBMSEventAdapterServiceDS;
import org.wso2.carbon.event.processor.manager.core.EventProcessorManagementService;
import org.wso2.carbon.event.processor.manager.core.EventPublisherManagementService;
import org.wso2.carbon.event.publisher.core.EventPublisherService;
import org.wso2.carbon.event.publisher.core.config.EventPublisherConfiguration;
import org.wso2.carbon.event.publisher.core.config.mapping.JSONOutputMapping;
import org.wso2.carbon.event.publisher.core.config.mapping.MapOutputMapping;
import org.wso2.carbon.event.publisher.core.exception.EventPublisherConfigurationException;
import org.wso2.carbon.event.publisher.core.internal.ds.EventPublisherServiceDS;
import org.wso2.carbon.event.publisher.stub.EventPublisherAdminServiceCallbackHandler;
import org.wso2.carbon.event.publisher.stub.EventPublisherAdminServiceStub;
import org.wso2.carbon.event.receiver.core.EventReceiverService;
import org.wso2.carbon.event.receiver.core.config.EventReceiverConfiguration;
import org.wso2.carbon.event.receiver.core.config.InputMapping;
import org.wso2.carbon.event.receiver.core.config.mapping.JSONInputMapping;
import org.wso2.carbon.event.receiver.core.config.mapping.WSO2EventInputMapping;
import org.wso2.carbon.event.receiver.core.exception.EventReceiverConfigurationException;
@ -59,7 +41,6 @@ import org.wso2.carbon.event.receiver.stub.EventReceiverAdminServiceCallbackHand
import org.wso2.carbon.event.receiver.stub.EventReceiverAdminServiceStub;
import org.wso2.carbon.event.receiver.stub.types.BasicInputAdapterPropertyDto;
import org.wso2.carbon.event.receiver.stub.types.EventReceiverConfigurationDto;
import org.wso2.carbon.event.receiver.stub.types.InputAdapterConfigurationDto;
import org.wso2.carbon.event.stream.core.EventStreamService;
import org.wso2.carbon.event.stream.core.exception.EventStreamConfigurationException;
import org.wso2.carbon.event.stream.stub.EventStreamAdminServiceStub;
@ -71,7 +52,6 @@ import org.wso2.carbon.user.api.UserStoreException;
import javax.validation.Valid;
import javax.ws.rs.*;
import javax.ws.rs.core.Response;
import java.io.IOException;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.HashMap;
@ -225,6 +205,7 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
String tenantDomain = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantDomain();
int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
try {
for (DeviceTypeEvent deviceTypeEvent : deviceTypeEvents) {
TransportType transportType = deviceTypeEvent.getTransportType();
@ -239,46 +220,101 @@ public class DeviceEventManagementServiceImpl implements DeviceEventManagementSe
log.error(errorMessage);
return Response.status(Response.Status.BAD_REQUEST).build();
}
// event stream
String streamName = DeviceMgtAPIUtils.getStreamDefinition(deviceType, tenantDomain, eventName);
publishStreamDefinitons(streamName, Constants.DEFAULT_STREAM_VERSION, eventAttributes);
AnalyticsArtifactsDeployer artifactsDeployer = new AnalyticsArtifactsDeployer();
List<Property> props = new ArrayList<>();
for (Attribute attribute : eventAttributes.getList()) {
props.add(new Property(attribute.getName(), attribute.getType().name()));
}
EventStreamData eventStreamData = new EventStreamData();
eventStreamData.setName(streamName);
eventStreamData.setVersion(Constants.DEFAULT_STREAM_VERSION);
eventStreamData.setMetaData(new MetaData(DEFAULT_DEVICE_ID_ATTRIBUTE, "STRING"));
eventStreamData.setPayloadData(props);
artifactsDeployer.deployEventStream(eventStreamData, tenantId);
// event receiver
String receiverName = getReceiverName(deviceType, tenantDomain, transportType, eventName);
publishEventReceivers(streamName, Constants.DEFAULT_STREAM_VERSION, transportType, tenantDomain,
isSharedWithAllTenants, deviceType, deviceTypeEvent.getEventTopicStructure(), receiverName);
EventReceiverData receiverData = new EventReceiverData();
receiverData.setName(receiverName);
receiverData.setStreamName(streamName);
receiverData.setStreamVersion(Constants.DEFAULT_STREAM_VERSION);
List<Property> propertyList = new ArrayList<>();
if (transportType == TransportType.MQTT) {
receiverData.setEventAdapterType(OAUTH_MQTT_ADAPTER_TYPE);
propertyList.add(new Property(MQTT_CONTENT_TRANSFORMER_TYPE, MQTT_CONTENT_TRANSFORMER));
propertyList.add(new Property(MQTT_CONTENT_VALIDATOR_TYPE, MQTT_CONTENT_VALIDATOR));
String topic;
if (!StringUtils.isEmpty(deviceTypeEvent.getEventTopicStructure())) {
if (isSharedWithAllTenants) {
topic = deviceTypeEvent.getEventTopicStructure().replace("${deviceId}", "+")
.replace("${deviceType}", deviceType)
.replace("${tenantDomain}", "+");
} else {
topic = deviceTypeEvent.getEventTopicStructure().replace("${deviceId}", "+")
.replace("${deviceType}", deviceType)
.replace("${tenantDomain}", tenantDomain);
}
} else {
if (isSharedWithAllTenants) {
topic = "+/" + deviceType + "/+/events";
} else {
topic = tenantDomain + "/" + deviceType + "/+/events";
}
}
propertyList.add(new Property("topic", topic));
receiverData.setCustomMappingType("json");
} else {
receiverData.setEventAdapterType(THRIFT_ADAPTER_TYPE);
propertyList.add(new Property("events.duplicated.in.cluster", "false"));
receiverData.setCustomMappingType("wso2event");
}
receiverData.setPropertyList(propertyList);
artifactsDeployer.deployEventReceiver(receiverData, tenantId);
if (!skipPersist) {
// rdbms event publisher
String rdbmsPublisherName = getPublisherName(deviceType, tenantDomain, eventName) + "_rdbms_publisher";
publishEventStore(streamName, Constants.DEFAULT_STREAM_VERSION, rdbmsPublisherName);
EventPublisherData eventPublisherData = new EventPublisherData();
eventPublisherData.setName(rdbmsPublisherName);
eventPublisherData.setStreamName(streamName);
eventPublisherData.setStreamVersion(Constants.DEFAULT_STREAM_VERSION);
eventPublisherData.setEventAdaptorType("rdbms");
eventPublisherData.setCustomMappingType("map");
List<Property> publisherProps = new ArrayList<>();
publisherProps.add(new Property("datasource.name", "EVENT_DB"));
publisherProps.add(new Property("table.name", "table_" + rdbmsPublisherName.replace(".", "")));
publisherProps.add(new Property("execution.mode", "insert"));
eventPublisherData.setPropertyList(publisherProps);
artifactsDeployer.deployEventPublisher(eventPublisherData, tenantId);
}
// web socket event publisher
String wsPublisherName = getPublisherName(deviceType, tenantDomain, eventName) + "_ws_publisher";
publishWebsocketPublisherDefinition(streamName, Constants.DEFAULT_STREAM_VERSION, wsPublisherName);
try {
PrivilegedCarbonContext.startTenantFlow();
PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantDomain(
MultitenantConstants.SUPER_TENANT_DOMAIN_NAME, true);
if (!MultitenantConstants.SUPER_TENANT_DOMAIN_NAME.equals(tenantDomain)) {
publishStreamDefinitons(streamName, Constants.DEFAULT_STREAM_VERSION, eventAttributes);
publishEventReceivers(streamName, Constants.DEFAULT_STREAM_VERSION, transportType, tenantDomain,
isSharedWithAllTenants, deviceType, deviceTypeEvent.getEventTopicStructure(), receiverName);
}
} finally {
PrivilegedCarbonContext.endTenantFlow();
}
EventPublisherData wsEventPublisherData = new EventPublisherData();
wsEventPublisherData.setName(wsPublisherName);
wsEventPublisherData.setStreamName(streamName);
wsEventPublisherData.setStreamVersion(Constants.DEFAULT_STREAM_VERSION);
wsEventPublisherData.setEventAdaptorType("websocket-local");
wsEventPublisherData.setCustomMappingType("json");
artifactsDeployer.deployEventPublisher(wsEventPublisherData, tenantId);
}
return Response.ok().build();
} catch (DeviceManagementException e) {
log.error("Failed to access device management service, tenantDomain: " + tenantDomain, e);
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
} catch (MalformedStreamDefinitionException e) {
log.error("Failed while creating stream definition, tenantDomain: " + tenantDomain, e);
} catch (EventStreamDeployerException e) {
log.error("Failed while deploying event stream definition, tenantDomain: " + tenantDomain, e);
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
} catch (EventStreamConfigurationException e) {
log.error("Failed while configuring stream definition, tenantDomain: " + tenantDomain, e);
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
} catch (EventPublisherConfigurationException e) {
log.error("Failed while configuring event publisher, tenantDomain: " + tenantDomain, e);
} catch (EventPublisherDeployerException e) {
log.error("Failed while deploying event publisher, tenantDomain: " + tenantDomain, e);
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
} catch (EventReceiverConfigurationException e) {
log.error("Failed while configuring event receiver, tenantDomain: " + tenantDomain, e);
} catch (EventReceiverDeployerException e) {
log.error("Failed while deploying event receiver, tenantDomain: " + tenantDomain, e);
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
}
}

@ -0,0 +1,99 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright (c) 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
~
~ Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
~ Version 2.0 (the "License"); you may not use this file except
~ in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing,
~ software distributed under the License is distributed on an
~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
~ KIND, either express or implied. See the License for the
~ specific language governing permissions and limitations
~ under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.wso2.carbon.devicemgt</groupId>
<artifactId>apimgt-extensions-feature</artifactId>
<version>5.0.21-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>io.entgra.device.mgt.core.apimgt.analytics.extension.feature</artifactId>
<packaging>pom</packaging>
<name>Entgra - APIM Analytics Extension Feature</name>
<url>http://entgra.io</url>
<dependencies>
<dependency>
<groupId>org.wso2.carbon.devicemgt</groupId>
<artifactId>io.entgra.device.mgt.core.apimgt.analytics.extension</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-resources-plugin</artifactId>
<version>2.6</version>
<executions>
<execution>
<id>copy-resources</id>
<phase>generate-resources</phase>
<goals>
<goal>copy-resources</goal>
</goals>
<configuration>
<outputDirectory>src/main/resources</outputDirectory>
<resources>
<resource>
<directory>resources</directory>
<includes>
<include>build.properties</include>
<include>p2.inf</include>
</includes>
</resource>
</resources>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.wso2.maven</groupId>
<artifactId>carbon-p2-plugin</artifactId>
<version>${carbon.p2.plugin.version}</version>
<executions>
<execution>
<id>p2-feature-generation</id>
<phase>package</phase>
<goals>
<goal>p2-feature-gen</goal>
</goals>
<configuration>
<id>io.entgra.device.mgt.core.apimgt.analytics.extension</id>
<propertiesFile>../../../features/etc/feature.properties</propertiesFile>
<adviceFile>
<properties>
<propertyDef>org.wso2.carbon.p2.category.type:server</propertyDef>
<propertyDef>org.eclipse.equinox.p2.type.group:true</propertyDef>
</properties>
</adviceFile>
<bundles>
<bundleDef>
org.wso2.carbon.devicemgt:io.entgra.device.mgt.core.apimgt.analytics.extension:${carbon.device.mgt.version}
</bundleDef>
</bundles>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

@ -0,0 +1,19 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright (c) 2023, Entgra (pvt) Ltd. (http://entgra.io)
~
~ All Rights Reserved.
~
~ Unauthorized copying of this file, via any medium is strictly prohibited.
~ Proprietary and confidential.
-->
<eventPublisher name="${name}" processing="enable"
statistics="disable" trace="disable" xmlns="http://wso2.org/carbon/eventpublisher">
<from streamName="${streamName}" version="${streamVersion}"/>
<mapping customMapping="disable" type="${customMappingType}"/>
<to eventAdapterType="${eventAdapterType}">
#foreach( $property in $properties )
<property name="${property.name}">${property.value}</property>
#end
</to>
</eventPublisher>

@ -0,0 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright (c) 2023, Entgra (pvt) Ltd. (http://entgra.io)
~
~ All Rights Reserved.
~
~ Unauthorized copying of this file, via any medium is strictly prohibited.
~ Proprietary and confidential.
-->
<eventReceiver name="${name}"
statistics="disable" trace="disable" xmlns="http://wso2.org/carbon/eventreceiver">
<from eventAdapterType="${eventAdapterType}">
#foreach( $property in $properties )
<property name="${property.name}">${property.value}</property>
#end
</from>
<mapping customMapping="disable" type="${customMappingType}"/>
<to
streamName="${streamName}" version="${streamVersion}"/>
</eventReceiver>

@ -0,0 +1,26 @@
{
"streamId": "$name:$version",
"name": "$name",
"version": "$version",
"nickName": "",
"description": "",
"metaData": [
{
"name": "${metaData.name}",
"type": "${metaData.type}"
}
],
"correlationData": [],
"payloadData": [
#foreach( $property in $properties )
{
"name": "${property.name}",
"type": "${property.value}"
},
#end
{
"name": "ts",
"type": "LONG"
}
]
}

@ -0,0 +1,2 @@
instructions.configure = \
org.eclipse.equinox.p2.touchpoint.natives.copy(source:${installFolder}/../features/io.entgra.device.mgt.core.apimgt.analytics.extension_${feature.version}/iot-analytics-templates/,target:${installFolder}/../../resources/,overwrite:true);\

@ -37,6 +37,7 @@
<module>org.wso2.carbon.apimgt.webapp.publisher.feature</module>
<module>org.wso2.carbon.apimgt.application.extension.feature</module>
<module>org.wso2.carbon.apimgt.keymgt.extension.feature</module>
<module>io.entgra.device.mgt.core.apimgt.analytics.extension.feature</module>
</modules>
</project>

@ -284,6 +284,11 @@
<artifactId>org.wso2.carbon.apimgt.application.extension</artifactId>
<version>${carbon.device.mgt.version}</version>
</dependency>
<dependency>
<groupId>org.wso2.carbon.devicemgt</groupId>
<artifactId>io.entgra.device.mgt.core.apimgt.analytics.extension</artifactId>
<version>${carbon.device.mgt.version}</version>
</dependency>
<dependency>
<groupId>org.wso2.carbon.devicemgt</groupId>
<artifactId>org.wso2.carbon.identity.jwt.client.extension</artifactId>

Loading…
Cancel
Save