Browse Source

SQS and SNS Plugins added

pull/455/head
mp-loki 9 years ago
committed by Andrew Shvayka
parent
commit
bdaee28d9e
  1. 2
      application/src/main/java/org/thingsboard/server/actors/rule/RuleActorMessageProcessor.java
  2. 80
      extensions/extension-sns/pom.xml
  3. 37
      extensions/extension-sns/src/assembly/extension.xml
  4. 31
      extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionMsg.java
  5. 37
      extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionPayload.java
  6. 45
      extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java
  7. 30
      extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginActionConfiguration.java
  8. 63
      extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java
  9. 79
      extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPlugin.java
  10. 30
      extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPluginConfiguration.java
  11. 30
      extensions/extension-sns/src/main/resources/SnsPluginDescriptor.json
  12. 34
      extensions/extension-sns/src/main/resources/SnsTopicActionDescriptor.json
  13. 80
      extensions/extention-sqs/pom.xml
  14. 37
      extensions/extention-sqs/src/assembly/extension.xml
  15. 31
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionMsg.java
  16. 39
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionPayload.java
  17. 49
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java
  18. 30
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginActionConfiguration.java
  19. 31
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionMsg.java
  20. 39
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionPayload.java
  21. 46
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java
  22. 32
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginActionConfiguration.java
  23. 84
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java
  24. 78
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPlugin.java
  25. 30
      extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPluginConfiguration.java
  26. 34
      extensions/extention-sqs/src/main/resources/SqsFifoQueueActionDescriptor.json
  27. 30
      extensions/extention-sqs/src/main/resources/SqsPluginDescriptor.json
  28. 41
      extensions/extention-sqs/src/main/resources/SqsStandardQueueActionDescriptor.json
  29. 69
      extensions/extention-sqs/src/test/java/org/thingsboard/server/extensions/sqs/SqsDemoClient.java
  30. 10
      extensions/extention-sqs/src/test/resources/logback.xml
  31. 2
      extensions/pom.xml
  32. 12
      pom.xml
  33. 18
      resume.bat
  34. 44
      tools/src/main/python/mqtt-send-telemetry.py

2
application/src/main/java/org/thingsboard/server/actors/rule/RuleActorMessageProcessor.java

@ -170,7 +170,7 @@ class RuleActorMessageProcessor extends ComponentMsgProcessor<RuleId> {
Optional<RuleToPluginMsg<?>> ruleToPluginMsgOptional = action.convert(ruleCtx, inMsg, inMsgMd);
if (ruleToPluginMsgOptional.isPresent()) {
RuleToPluginMsg<?> ruleToPluginMsg = ruleToPluginMsgOptional.get();
logger.debug("[{}] Device msg is converter to: {}", entityId, ruleToPluginMsg);
logger.debug("[{}] Device msg is converted to: {}", entityId, ruleToPluginMsg);
context.parent().tell(new RuleToPluginMsgWrapper(pluginTenantId, pluginId, tenantId, entityId, ruleToPluginMsg), context.self());
if (action.isOneWayAction()) {
pushToNextRule(context, msg.getCtx(), RuleEngineError.NO_TWO_WAY_ACTIONS);

80
extensions/extension-sns/pom.xml

@ -0,0 +1,80 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Copyright © 2016-2017 The Thingsboard Authors
Licensed 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>extensions</artifactId>
<groupId>org.thingsboard</groupId>
<version>1.4.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>org.thingsboard.extensions</groupId>
<artifactId>extension-sns</artifactId>
<packaging>jar</packaging>
<name>Thingsboard Server SNS Extension</name>
<url>http://thingsboard.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<main.dir>${basedir}/../..</main.dir>
<aws.sdk.version>1.11.229</aws.sdk.version>
</properties>
<dependencies>
<dependency>
<groupId>org.thingsboard</groupId>
<artifactId>extensions-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sns</artifactId>
<version>${aws.sdk.version}</version>
</dependency>
<dependency>
<groupId>org.thingsboard</groupId>
<artifactId>extensions-core</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptors>
<descriptor>src/assembly/extension.xml</descriptor>
</descriptors>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

37
extensions/extension-sns/src/assembly/extension.xml

@ -0,0 +1,37 @@
<!--
Copyright © 2016-2017 The Thingsboard Authors
Licensed 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.
-->
<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.0.0 http://maven.apache.org/xsd/assembly-2.0.0.xsd">
<id>extension</id>
<formats>
<format>jar</format>
</formats>
<includeBaseDirectory>false</includeBaseDirectory>
<dependencySets>
<dependencySet>
<outputDirectory>/</outputDirectory>
<useProjectArtifact>true</useProjectArtifact>
<unpack>true</unpack>
<scope>runtime</scope>
<excludes>
</excludes>
</dependencySet>
</dependencySets>
</assembly>

31
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionMsg.java

@ -0,0 +1,31 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sns.action;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.extensions.api.plugins.msg.AbstractRuleToPluginMsg;
/**
* Created by Valerii Sosliuk on 11/15/2017.
*/
public class SnsTopicActionMsg extends AbstractRuleToPluginMsg<SnsTopicActionPayload> {
public SnsTopicActionMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, SnsTopicActionPayload payload) {
super(tenantId, customerId, deviceId, payload);
}
}

37
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionPayload.java

@ -0,0 +1,37 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sns.action;
import lombok.Builder;
import lombok.Data;
import org.thingsboard.server.common.msg.session.MsgType;
import java.io.Serializable;
/**
* Created by Valerii Sosliuk on 11/15/2017.
*/
@Data
@Builder
public class SnsTopicActionPayload implements Serializable {
private final String topicArn;
private final String msgBody;
private final Integer requestId;
private final MsgType msgType;
private final boolean sync;
}

45
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java

@ -0,0 +1,45 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sns.action;
import org.thingsboard.server.common.msg.device.ToDeviceActorMsg;
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg;
import org.thingsboard.server.extensions.api.component.Action;
import org.thingsboard.server.extensions.api.plugins.msg.RuleToPluginMsg;
import org.thingsboard.server.extensions.api.rules.RuleContext;
import org.thingsboard.server.extensions.core.action.template.AbstractTemplatePluginAction;
import java.util.Optional;
/**
* Created by Valerii Sosliuk on 11/15/2017.
*/
@Action(name = "SNS Topic Action", descriptor = "SnsTopicActionDescriptor.json", configuration = SnsTopicPluginActionConfiguration.class)
public class SnsTopicPluginAction extends AbstractTemplatePluginAction<SnsTopicPluginActionConfiguration> {
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, ToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
SnsTopicActionPayload.SnsTopicActionPayloadBuilder builder = SnsTopicActionPayload.builder();
builder.msgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.topicArn(configuration.getTopicArn());
builder.msgBody(getMsgBody(ctx, msg));
return Optional.of(new SnsTopicActionMsg(msg.getTenantId(),
msg.getCustomerId(),
msg.getDeviceId(),
builder.build()));
}
}

30
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginActionConfiguration.java

@ -0,0 +1,30 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sns.action;
import lombok.Data;
import org.thingsboard.server.extensions.core.action.template.TemplateActionConfiguration;
/**
* Created by Valerii Sosliuk on 11/15/2017.
*/
@Data
public class SnsTopicPluginActionConfiguration implements TemplateActionConfiguration {
private String topicArn;
private String template;
private boolean sync;
}

63
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java

@ -0,0 +1,63 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sns.plugin;
import com.amazonaws.services.sns.AmazonSNS;
import com.amazonaws.services.sns.model.PublishRequest;
import com.amazonaws.services.sns.model.PublishResult;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.id.RuleId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.core.BasicStatusCodeResponse;
import org.thingsboard.server.extensions.api.plugins.PluginContext;
import org.thingsboard.server.extensions.api.plugins.handlers.RuleMsgHandler;
import org.thingsboard.server.extensions.api.plugins.msg.ResponsePluginToRuleMsg;
import org.thingsboard.server.extensions.api.plugins.msg.RuleToPluginMsg;
import org.thingsboard.server.extensions.api.rules.RuleException;
import org.thingsboard.server.extensions.sns.action.SnsTopicActionMsg;
import org.thingsboard.server.extensions.sns.action.SnsTopicActionPayload;
/**
* Created by Valerii Sosliuk on 11/6/2017.
*/
@RequiredArgsConstructor
@Slf4j
public class SnsMessageHandler implements RuleMsgHandler {
private final AmazonSNS sns;
@Override
public void process(PluginContext ctx, TenantId tenantId, RuleId ruleId, RuleToPluginMsg<?> msg) throws RuleException {
if (msg instanceof SnsTopicActionMsg) {
SnsTopicActionPayload payload = ((SnsTopicActionMsg) msg).getPayload();
PublishRequest publishRequest = new PublishRequest()
.withTopicArn(payload.getTopicArn())
.withMessage(payload.getMsgBody());
sns.publish(publishRequest);
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
}
return;
}
throw new RuleException("Unsupported message type " + msg.getClass().getName() + "!");
}
}

79
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPlugin.java

@ -0,0 +1,79 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sns.plugin;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.sns.AmazonSNS;
import com.amazonaws.services.sns.AmazonSNSClient;
import org.thingsboard.server.extensions.api.component.Plugin;
import org.thingsboard.server.extensions.api.plugins.AbstractPlugin;
import org.thingsboard.server.extensions.api.plugins.PluginContext;
import org.thingsboard.server.extensions.api.plugins.handlers.RuleMsgHandler;
import org.thingsboard.server.extensions.sns.action.SnsTopicPluginAction;
/**
* Created by Valerii Sosliuk on 11/15/2017.
*/
@Plugin(name = "SNS Plugin", actions = {SnsTopicPluginAction.class},
descriptor = "SnsPluginDescriptor.json", configuration = SnsPluginConfiguration.class)
public class SnsPlugin extends AbstractPlugin<SnsPluginConfiguration> {
private SnsMessageHandler snsMessageHandler;
private SnsPluginConfiguration configuration;
@Override
public void init(SnsPluginConfiguration configuration) {
this.configuration = configuration;
init();
}
private void init() {
AWSCredentials awsCredentials = new BasicAWSCredentials(configuration.getAccessKeyId(), configuration.getSecretAccessKey());
AWSStaticCredentialsProvider credProvider = new AWSStaticCredentialsProvider(awsCredentials);
AmazonSNS sns = AmazonSNSClient.builder()
.withCredentials(credProvider)
.withRegion(configuration.getRegion())
.build();
this.snsMessageHandler = new SnsMessageHandler(sns);
}
private void destroy() {
this.snsMessageHandler = null;
}
@Override
protected RuleMsgHandler getRuleMsgHandler() {
return snsMessageHandler;
}
@Override
public void resume(PluginContext ctx) {
init();
}
@Override
public void suspend(PluginContext ctx) {
destroy();
}
@Override
public void stop(PluginContext ctx) {
destroy();
}
}

30
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPluginConfiguration.java

@ -0,0 +1,30 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sns.plugin;
import lombok.Data;
/**
* Created by Valerii Sosliuk on 11/5/2017.
*/
@Data
public class SnsPluginConfiguration {
private String accessKeyId;
private String secretAccessKey;
private String region;
}

30
extensions/extension-sns/src/main/resources/SnsPluginDescriptor.json

@ -0,0 +1,30 @@
{
"schema": {
"title": "SNS Plugin Configuration",
"type": "object",
"properties": {
"accessKeyId": {
"title": "Access Key ID",
"type": "string"
},
"secretAccessKey": {
"title": "Secret Access Key",
"type": "string"
},
"region": {
"title": "Region",
"type": "string"
}
},
"required": [
"accessKeyId",
"secretAccessKey",
"region"
]
},
"form": [
"accessKeyId",
"secretAccessKey",
"region"
]
}

34
extensions/extension-sns/src/main/resources/SnsTopicActionDescriptor.json

@ -0,0 +1,34 @@
{
"schema": {
"title": "SNS Topic Action Configuration",
"type": "object",
"properties": {
"sync": {
"title": "Requires delivery confirmation",
"type": "boolean"
},
"topicArn": {
"title": "Topic ARN",
"type": "string"
},
"template": {
"title": "Body Template",
"type": "string"
}
},
"required": [
"sync",
"topicArn",
"template"
]
},
"form": [
"sync",
"topicArn",
{
"key": "template",
"type": "textarea",
"rows": 5
}
]
}

80
extensions/extention-sqs/pom.xml

@ -0,0 +1,80 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Copyright © 2016-2017 The Thingsboard Authors
Licensed 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>extensions</artifactId>
<groupId>org.thingsboard</groupId>
<version>1.4.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>org.thingsboard.extensions</groupId>
<artifactId>extension-sqs</artifactId>
<packaging>jar</packaging>
<name>Thingsboard Server SQS Extension</name>
<url>http://thingsboard.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<main.dir>${basedir}/../..</main.dir>
<aws.sdk.version>1.11.229</aws.sdk.version>
</properties>
<dependencies>
<dependency>
<groupId>org.thingsboard</groupId>
<artifactId>extensions-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sqs</artifactId>
<version>${aws.sdk.version}</version>
</dependency>
<dependency>
<groupId>org.thingsboard</groupId>
<artifactId>extensions-core</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptors>
<descriptor>src/assembly/extension.xml</descriptor>
</descriptors>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

37
extensions/extention-sqs/src/assembly/extension.xml

@ -0,0 +1,37 @@
<!--
Copyright © 2016-2017 The Thingsboard Authors
Licensed 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.
-->
<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.0.0 http://maven.apache.org/xsd/assembly-2.0.0.xsd">
<id>extension</id>
<formats>
<format>jar</format>
</formats>
<includeBaseDirectory>false</includeBaseDirectory>
<dependencySets>
<dependencySet>
<outputDirectory>/</outputDirectory>
<useProjectArtifact>true</useProjectArtifact>
<unpack>true</unpack>
<scope>runtime</scope>
<excludes>
</excludes>
</dependencySet>
</dependencySets>
</assembly>

31
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionMsg.java

@ -0,0 +1,31 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.fifo;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.extensions.api.plugins.msg.AbstractRuleToPluginMsg;
/**
* Created by Valerii Sosliuk on 11/10/2017.
*/
public class SqsFifoQueueActionMsg extends AbstractRuleToPluginMsg<SqsFifoQueueActionPayload> {
public SqsFifoQueueActionMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, SqsFifoQueueActionPayload payload) {
super(tenantId, customerId, deviceId, payload);
}
}

39
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionPayload.java

@ -0,0 +1,39 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.fifo;
import lombok.Builder;
import lombok.Data;
import org.thingsboard.server.common.msg.session.MsgType;
import java.io.Serializable;
/**
* Created by Valerii Sosliuk on 11/10/2017.
*/
@Data
@Builder
public class SqsFifoQueueActionPayload implements Serializable {
private final String queue;
private final String msgBody;
private final String deviceId;
private final Integer requestId;
private final MsgType msgType;
private final boolean sync;
}

49
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java

@ -0,0 +1,49 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.fifo;
import org.thingsboard.server.common.msg.device.ToDeviceActorMsg;
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg;
import org.thingsboard.server.extensions.api.component.Action;
import org.thingsboard.server.extensions.api.plugins.msg.RuleToPluginMsg;
import org.thingsboard.server.extensions.api.rules.RuleContext;
import org.thingsboard.server.extensions.core.action.template.AbstractTemplatePluginAction;
import org.thingsboard.server.extensions.sqs.action.standard.SqsStandardQueueActionMsg;
import org.thingsboard.server.extensions.sqs.action.standard.SqsStandardQueueActionPayload;
import org.thingsboard.server.extensions.sqs.action.standard.SqsStandardQueuePluginActionConfiguration;
import java.util.Optional;
/**
* Created by Valerii Sosliuk on 11/5/2017.
*/
@Action(name = "SQS Fifo Queue Action", descriptor = "SqsFifoQueueActionDescriptor.json", configuration = SqsFifoQueuePluginActionConfiguration.class)
public class SqsFifoQueuePluginAction extends AbstractTemplatePluginAction<SqsFifoQueuePluginActionConfiguration> {
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, ToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
SqsFifoQueueActionPayload.SqsFifoQueueActionPayloadBuilder builder = SqsFifoQueueActionPayload.builder();
builder.msgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.queue(configuration.getQueue());
builder.deviceId(msg.getDeviceId().toString());
builder.msgBody(getMsgBody(ctx, msg));
return Optional.of(new SqsFifoQueueActionMsg(msg.getTenantId(),
msg.getCustomerId(),
msg.getDeviceId(),
builder.build()));
}
}

30
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginActionConfiguration.java

@ -0,0 +1,30 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.fifo;
import lombok.Data;
import org.thingsboard.server.extensions.core.action.template.TemplateActionConfiguration;
/**
* Created by Valerii Sosliuk on 11/10/2017.
*/
@Data
public class SqsFifoQueuePluginActionConfiguration implements TemplateActionConfiguration {
private String queue;
private String template;
private boolean sync;
}

31
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionMsg.java

@ -0,0 +1,31 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.standard;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.extensions.api.plugins.msg.AbstractRuleToPluginMsg;
/**
* Created by Valerii Sosliuk on 11/6/2017.
*/
public class SqsStandardQueueActionMsg extends AbstractRuleToPluginMsg<SqsStandardQueueActionPayload> {
public SqsStandardQueueActionMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, SqsStandardQueueActionPayload payload) {
super(tenantId, customerId, deviceId, payload);
}
}

39
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionPayload.java

@ -0,0 +1,39 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.standard;
import lombok.Builder;
import lombok.Data;
import org.thingsboard.server.common.msg.session.MsgType;
import java.io.Serializable;
/**
* Created by Valerii Sosliuk on 11/6/2017.
*/
@Data
@Builder
public class SqsStandardQueueActionPayload implements Serializable {
private final String queue;
private final String msgBody;
private final int delaySeconds;
private final Integer requestId;
private final MsgType msgType;
private final boolean sync;
}

46
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java

@ -0,0 +1,46 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.standard;
import org.thingsboard.server.common.msg.device.ToDeviceActorMsg;
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg;
import org.thingsboard.server.extensions.api.component.Action;
import org.thingsboard.server.extensions.api.plugins.msg.RuleToPluginMsg;
import org.thingsboard.server.extensions.api.rules.RuleContext;
import org.thingsboard.server.extensions.core.action.template.AbstractTemplatePluginAction;
import java.util.Optional;
/**
* Created by Valerii Sosliuk on 11/5/2017.
*/
@Action(name = "SQS Standard Queue Action", descriptor = "SqsStandardQueueActionDescriptor.json", configuration = SqsStandardQueuePluginActionConfiguration.class)
public class SqsStandardQueuePluginAction extends AbstractTemplatePluginAction<SqsStandardQueuePluginActionConfiguration> {
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, ToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
SqsStandardQueueActionPayload.SqsStandardQueueActionPayloadBuilder builder = SqsStandardQueueActionPayload.builder();
builder.msgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.queue(configuration.getQueue());
builder.delaySeconds(configuration.getDelaySeconds());
builder.msgBody(getMsgBody(ctx, msg));
return Optional.of(new SqsStandardQueueActionMsg(msg.getTenantId(),
msg.getCustomerId(),
msg.getDeviceId(),
builder.build()));
}
}

32
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginActionConfiguration.java

@ -0,0 +1,32 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.action.standard;
import lombok.Data;
import org.thingsboard.server.extensions.core.action.template.TemplateActionConfiguration;
/**
* Created by Valerii Sosliuk on 11/6/2017.
*/
@Data
public class SqsStandardQueuePluginActionConfiguration implements TemplateActionConfiguration {
private String queue;
private int delaySeconds;
private boolean sync;
private String template;
}

84
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java

@ -0,0 +1,84 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.plugin;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageResult;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.id.RuleId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.core.BasicStatusCodeResponse;
import org.thingsboard.server.extensions.api.plugins.PluginContext;
import org.thingsboard.server.extensions.api.plugins.handlers.RuleMsgHandler;
import org.thingsboard.server.extensions.api.plugins.msg.AbstractRuleToPluginMsg;
import org.thingsboard.server.extensions.api.plugins.msg.ResponsePluginToRuleMsg;
import org.thingsboard.server.extensions.api.plugins.msg.RuleToPluginMsg;
import org.thingsboard.server.extensions.api.rules.RuleException;
import org.thingsboard.server.extensions.sqs.action.fifo.SqsFifoQueueActionMsg;
import org.thingsboard.server.extensions.sqs.action.fifo.SqsFifoQueueActionPayload;
import org.thingsboard.server.extensions.sqs.action.standard.SqsStandardQueueActionMsg;
import org.thingsboard.server.extensions.sqs.action.standard.SqsStandardQueueActionPayload;
/**
* Created by Valerii Sosliuk on 11/15/2017.
*/
@RequiredArgsConstructor
@Slf4j
public class SqsMessageHandler implements RuleMsgHandler {
private final AmazonSQS sqs;
@Override
public void process(PluginContext ctx, TenantId tenantId, RuleId ruleId, RuleToPluginMsg<?> msg) throws RuleException {
if (msg instanceof SqsStandardQueueActionMsg) {
sendMessageToStandardQueue(ctx, tenantId, ruleId, msg);
return;
}
if (msg instanceof SqsFifoQueueActionMsg) {
sendMessageToFifoQueue(ctx, tenantId, ruleId, msg);
return;
}
throw new RuleException("Unsupported message type " + msg.getClass().getName() + "!");
}
private void sendMessageToStandardQueue(PluginContext ctx, TenantId tenantId, RuleId ruleId, RuleToPluginMsg<?> msg) {
SqsStandardQueueActionPayload payload = ((SqsStandardQueueActionMsg) msg).getPayload();
SendMessageRequest sendMsgRequest = new SendMessageRequest()
.withDelaySeconds(payload.getDelaySeconds())
.withQueueUrl(payload.getQueue())
.withMessageBody(payload.getMsgBody());
sqs.sendMessage(sendMsgRequest);
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
}
}
private void sendMessageToFifoQueue(PluginContext ctx, TenantId tenantId, RuleId ruleId, RuleToPluginMsg<?> msg) {
SqsFifoQueueActionPayload payload = ((SqsFifoQueueActionMsg) msg).getPayload();
SendMessageRequest sendMsgRequest = new SendMessageRequest()
.withQueueUrl(payload.getQueue())
.withMessageBody(payload.getMsgBody())
.withMessageGroupId(payload.getDeviceId());
sqs.sendMessage(sendMsgRequest);
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
}
}
}

78
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPlugin.java

@ -0,0 +1,78 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.plugin;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import org.thingsboard.server.extensions.api.component.Plugin;
import org.thingsboard.server.extensions.api.plugins.AbstractPlugin;
import org.thingsboard.server.extensions.api.plugins.PluginContext;
import org.thingsboard.server.extensions.api.plugins.handlers.RuleMsgHandler;
import org.thingsboard.server.extensions.sqs.action.fifo.SqsFifoQueuePluginAction;
import org.thingsboard.server.extensions.sqs.action.standard.SqsStandardQueuePluginAction;
/**
* Created by Valerii Sosliuk on 11/6/2017.
*/
@Plugin(name = "SQS Plugin", actions = {SqsStandardQueuePluginAction.class, SqsFifoQueuePluginAction.class},
descriptor = "SqsPluginDescriptor.json", configuration = SqsPluginConfiguration.class)
public class SqsPlugin extends AbstractPlugin<SqsPluginConfiguration> {
private SqsMessageHandler sqsMessageHandler;
private SqsPluginConfiguration configuration;
@Override
public void init(SqsPluginConfiguration configuration) {
this.configuration = configuration;
init();
}
private void init() {
AWSCredentials awsCredentials = new BasicAWSCredentials(configuration.getAccessKeyId(), configuration.getSecretAccessKey());
AmazonSQS sqs = AmazonSQSClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(awsCredentials))
.withRegion(Regions.fromName(configuration.getRegion())).build();
this.sqsMessageHandler = new SqsMessageHandler(sqs);
}
private void destroy() {
this.sqsMessageHandler = null;
}
@Override
protected RuleMsgHandler getRuleMsgHandler() {
return sqsMessageHandler;
}
@Override
public void resume(PluginContext ctx) {
init();
}
@Override
public void suspend(PluginContext ctx) {
destroy();
}
@Override
public void stop(PluginContext ctx) {
destroy();
}
}

30
extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPluginConfiguration.java

@ -0,0 +1,30 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs.plugin;
import lombok.Data;
/**
* Created by Valerii Sosliuk on 11/5/2017.
*/
@Data
public class SqsPluginConfiguration {
private String accessKeyId;
private String secretAccessKey;
private String region;
}

34
extensions/extention-sqs/src/main/resources/SqsFifoQueueActionDescriptor.json

@ -0,0 +1,34 @@
{
"schema": {
"title": "SQS FIFO Queue Action Configuration",
"type": "object",
"properties": {
"sync": {
"title": "Requires delivery confirmation",
"type": "boolean"
},
"queue": {
"title": "Queue URL",
"type": "string"
},
"template": {
"title": "Body Template",
"type": "string"
}
},
"required": [
"sync",
"queue",
"template"
]
},
"form": [
"sync",
"queue",
{
"key": "template",
"type": "textarea",
"rows": 5
}
]
}

30
extensions/extention-sqs/src/main/resources/SqsPluginDescriptor.json

@ -0,0 +1,30 @@
{
"schema": {
"title": "SQS Plugin Configuration",
"type": "object",
"properties": {
"accessKeyId": {
"title": "Access Key ID",
"type": "string"
},
"secretAccessKey": {
"title": "Secret Access Key",
"type": "string"
},
"region": {
"title": "Region",
"type": "string"
}
},
"required": [
"accessKeyId",
"secretAccessKey",
"region"
]
},
"form": [
"accessKeyId",
"secretAccessKey",
"region"
]
}

41
extensions/extention-sqs/src/main/resources/SqsStandardQueueActionDescriptor.json

@ -0,0 +1,41 @@
{
"schema": {
"title": "SQS Standard Queue Action Configuration",
"type": "object",
"properties": {
"sync": {
"title": "Requires delivery confirmation",
"type": "boolean"
},
"queue": {
"title": "Queue URL",
"type": "string"
},
"delaySeconds": {
"title": "Delay Seconds",
"type": "integer",
"default": 0
},
"template": {
"title": "Body Template",
"type": "string"
}
},
"required": [
"sync",
"queue",
"delaySeconds",
"template"
]
},
"form": [
"sync",
"queue",
"delaySeconds",
{
"key": "template",
"type": "textarea",
"rows": 5
}
]
}

69
extensions/extention-sqs/src/test/java/org/thingsboard/server/extensions/sqs/SqsDemoClient.java

@ -0,0 +1,69 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* Licensed 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 org.thingsboard.server.extensions.sqs;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.amazonaws.services.sqs.model.DeleteMessageRequest;
import com.amazonaws.services.sqs.model.Message;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
/**
* Created by Valerii Sosliuk on 11/10/2017.
*/
@Slf4j
public class SqsDemoClient {
private static final String ACCESS_KEY_ID = "$ACCES_KEY_ID";
private static final String SECRET_ACCESS_KEY = "$SECRET_ACCESS_KEY";
private static final String QUEUE_URL = "$QUEUE_URL";
private static final String REGION = "us-east-1";
public static void main(String[] args) {
log.info("Starting SQS Demo Clinent...");
AWSCredentials awsCredentials = new BasicAWSCredentials(ACCESS_KEY_ID, SECRET_ACCESS_KEY);
AmazonSQS sqs = AmazonSQSClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(awsCredentials))
.withRegion(Regions.fromName(REGION)).build();
SqsDemoClient client = new SqsDemoClient();
client.pollMessages(sqs);
}
private void pollMessages(AmazonSQS sqs) {
log.info("Polling messages");
while (true) {
List<Message> messages = sqs.receiveMessage(QUEUE_URL).getMessages();
messages.forEach(m -> {
log.info("Message Received: " + m.getBody());
System.out.println(m.getBody());
DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest(QUEUE_URL, m.getReceiptHandle());
sqs.deleteMessage(deleteMessageRequest);
});
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
e.printStackTrace();
}
}
}
}

10
extensions/extention-sqs/src/test/resources/logback.xml

@ -0,0 +1,10 @@
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<root level="INFO">
<appender-ref ref="STDOUT"/>
</root>
</configuration>

2
extensions/pom.xml

@ -39,6 +39,8 @@
<module>extension-rest-api-call</module>
<module>extension-kafka</module>
<module>extension-mqtt</module>
<module>extention-sqs</module>
<module>extension-sns</module>
</modules>
</project>

12
pom.xml

@ -350,6 +350,18 @@
<classifier>extension</classifier>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.thingsboard.extensions</groupId>
<artifactId>extension-sqs</artifactId>
<classifier>extension</classifier>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.thingsboard.extensions</groupId>
<artifactId>extension-sns</artifactId>
<classifier>extension</classifier>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.thingsboard.common</groupId>
<artifactId>data</artifactId>

18
resume.bat

@ -0,0 +1,18 @@
@REM
@REM Copyright © 2016-2017 The Thingsboard Authors
@REM
@REM Licensed under the Apache License, Version 2.0 (the "License");
@REM you may not use this file except in compliance with the License.
@REM You may obtain a copy of the License at
@REM
@REM http://www.apache.org/licenses/LICENSE-2.0
@REM
@REM Unless required by applicable law or agreed to in writing, software
@REM distributed under the License is distributed on an "AS IS" BASIS,
@REM WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@REM See the License for the specific language governing permissions and
@REM limitations under the License.
@REM
mvn clean install -rf :application

44
tools/src/main/python/mqtt-send-telemetry.py

@ -0,0 +1,44 @@
#
#
# Licensed 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.
#
import paho.mqtt.client as mqtt
from time import sleep
import random
broker="test.mosquitto.org"
topic_pub='v1/devices/me/telemetry'
client = mqtt.Client()
client.username_pw_set("qyA3gP50SpGwfwyNGyi7")
client.connect('127.0.0.1', 1883, 1)
for i in range(100):
x = random.randrange(20, 100)
print x
msg = '{"windSpeed":"'+ str(x) + '"}'
client.publish(topic_pub, msg)
sleep(0.1)
#while True:
# val3 = random.uniform(0, 5)
# val4 = random.uniform(-3, 3)#
#
# msg = '{"key3": '+ str(val3) +', "key4": ' + str(val4) + '}'
#
# print('Message: ' + msg)
# client.publish(topic_pub, msg)
#
# sleep(0.5)
Loading…
Cancel
Save