From bdaee28d9e39c91626235418e3f4013d0aafbf5d Mon Sep 17 00:00:00 2001 From: mp-loki Date: Tue, 7 Nov 2017 19:08:06 -0500 Subject: [PATCH] SQS and SNS Plugins added --- .../rule/RuleActorMessageProcessor.java | 2 +- extensions/extension-sns/pom.xml | 80 ++++++++++++++++++ .../extension-sns/src/assembly/extension.xml | 37 ++++++++ .../sns/action/SnsTopicActionMsg.java | 31 +++++++ .../sns/action/SnsTopicActionPayload.java | 37 ++++++++ .../sns/action/SnsTopicPluginAction.java | 45 ++++++++++ .../SnsTopicPluginActionConfiguration.java | 30 +++++++ .../sns/plugin/SnsMessageHandler.java | 63 ++++++++++++++ .../extensions/sns/plugin/SnsPlugin.java | 79 +++++++++++++++++ .../sns/plugin/SnsPluginConfiguration.java | 30 +++++++ .../main/resources/SnsPluginDescriptor.json | 30 +++++++ .../resources/SnsTopicActionDescriptor.json | 34 ++++++++ extensions/extention-sqs/pom.xml | 80 ++++++++++++++++++ .../extention-sqs/src/assembly/extension.xml | 37 ++++++++ .../action/fifo/SqsFifoQueueActionMsg.java | 31 +++++++ .../fifo/SqsFifoQueueActionPayload.java | 39 +++++++++ .../action/fifo/SqsFifoQueuePluginAction.java | 49 +++++++++++ ...SqsFifoQueuePluginActionConfiguration.java | 30 +++++++ .../standard/SqsStandardQueueActionMsg.java | 31 +++++++ .../SqsStandardQueueActionPayload.java | 39 +++++++++ .../SqsStandardQueuePluginAction.java | 46 ++++++++++ ...tandardQueuePluginActionConfiguration.java | 32 +++++++ .../sqs/plugin/SqsMessageHandler.java | 84 +++++++++++++++++++ .../extensions/sqs/plugin/SqsPlugin.java | 78 +++++++++++++++++ .../sqs/plugin/SqsPluginConfiguration.java | 30 +++++++ .../SqsFifoQueueActionDescriptor.json | 34 ++++++++ .../main/resources/SqsPluginDescriptor.json | 30 +++++++ .../SqsStandardQueueActionDescriptor.json | 41 +++++++++ .../server/extensions/sqs/SqsDemoClient.java | 69 +++++++++++++++ .../src/test/resources/logback.xml | 10 +++ extensions/pom.xml | 2 + pom.xml | 12 +++ resume.bat | 18 ++++ tools/src/main/python/mqtt-send-telemetry.py | 44 ++++++++++ 34 files changed, 1363 insertions(+), 1 deletion(-) create mode 100644 extensions/extension-sns/pom.xml create mode 100644 extensions/extension-sns/src/assembly/extension.xml create mode 100644 extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionMsg.java create mode 100644 extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionPayload.java create mode 100644 extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java create mode 100644 extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginActionConfiguration.java create mode 100644 extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java create mode 100644 extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPlugin.java create mode 100644 extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPluginConfiguration.java create mode 100644 extensions/extension-sns/src/main/resources/SnsPluginDescriptor.json create mode 100644 extensions/extension-sns/src/main/resources/SnsTopicActionDescriptor.json create mode 100644 extensions/extention-sqs/pom.xml create mode 100644 extensions/extention-sqs/src/assembly/extension.xml create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionMsg.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionPayload.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginActionConfiguration.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionMsg.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionPayload.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginActionConfiguration.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPlugin.java create mode 100644 extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPluginConfiguration.java create mode 100644 extensions/extention-sqs/src/main/resources/SqsFifoQueueActionDescriptor.json create mode 100644 extensions/extention-sqs/src/main/resources/SqsPluginDescriptor.json create mode 100644 extensions/extention-sqs/src/main/resources/SqsStandardQueueActionDescriptor.json create mode 100644 extensions/extention-sqs/src/test/java/org/thingsboard/server/extensions/sqs/SqsDemoClient.java create mode 100644 extensions/extention-sqs/src/test/resources/logback.xml create mode 100644 resume.bat create mode 100644 tools/src/main/python/mqtt-send-telemetry.py diff --git a/application/src/main/java/org/thingsboard/server/actors/rule/RuleActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/rule/RuleActorMessageProcessor.java index c695ec8316..7a7e3ff1ab 100644 --- a/application/src/main/java/org/thingsboard/server/actors/rule/RuleActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/rule/RuleActorMessageProcessor.java @@ -170,7 +170,7 @@ class RuleActorMessageProcessor extends ComponentMsgProcessor { Optional> 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); diff --git a/extensions/extension-sns/pom.xml b/extensions/extension-sns/pom.xml new file mode 100644 index 0000000000..37f8fac57c --- /dev/null +++ b/extensions/extension-sns/pom.xml @@ -0,0 +1,80 @@ + + + + + extensions + org.thingsboard + 1.4.0-SNAPSHOT + + 4.0.0 + org.thingsboard.extensions + extension-sns + jar + + Thingsboard Server SNS Extension + http://thingsboard.org + + + UTF-8 + ${basedir}/../.. + 1.11.229 + + + + + org.thingsboard + extensions-api + provided + + + com.amazonaws + aws-java-sdk-sns + ${aws.sdk.version} + + + org.thingsboard + extensions-core + + + + + + + maven-assembly-plugin + + + src/assembly/extension.xml + + + + + make-assembly + package + + single + + + + + + + + \ No newline at end of file diff --git a/extensions/extension-sns/src/assembly/extension.xml b/extensions/extension-sns/src/assembly/extension.xml new file mode 100644 index 0000000000..2395d5551a --- /dev/null +++ b/extensions/extension-sns/src/assembly/extension.xml @@ -0,0 +1,37 @@ + + + extension + + jar + + false + + + / + true + true + runtime + + + + + + \ No newline at end of file diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionMsg.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionMsg.java new file mode 100644 index 0000000000..b64a22bdba --- /dev/null +++ b/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 { + + public SnsTopicActionMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, SnsTopicActionPayload payload) { + super(tenantId, customerId, deviceId, payload); + } +} diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionPayload.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicActionPayload.java new file mode 100644 index 0000000000..a1bf89a248 --- /dev/null +++ b/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; +} diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java new file mode 100644 index 0000000000..79de23acd4 --- /dev/null +++ b/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 { + + @Override + protected Optional 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())); + } +} diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginActionConfiguration.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginActionConfiguration.java new file mode 100644 index 0000000000..468c3ddbc3 --- /dev/null +++ b/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; +} diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java new file mode 100644 index 0000000000..c7fd99c8be --- /dev/null +++ b/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() + "!"); + + } + +} diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPlugin.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPlugin.java new file mode 100644 index 0000000000..86f5de0331 --- /dev/null +++ b/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 { + + 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(); + } +} diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPluginConfiguration.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsPluginConfiguration.java new file mode 100644 index 0000000000..dee78f3f57 --- /dev/null +++ b/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; + +} diff --git a/extensions/extension-sns/src/main/resources/SnsPluginDescriptor.json b/extensions/extension-sns/src/main/resources/SnsPluginDescriptor.json new file mode 100644 index 0000000000..04e23e75bc --- /dev/null +++ b/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" + ] +} \ No newline at end of file diff --git a/extensions/extension-sns/src/main/resources/SnsTopicActionDescriptor.json b/extensions/extension-sns/src/main/resources/SnsTopicActionDescriptor.json new file mode 100644 index 0000000000..a8a2793885 --- /dev/null +++ b/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 + } + ] +} \ No newline at end of file diff --git a/extensions/extention-sqs/pom.xml b/extensions/extention-sqs/pom.xml new file mode 100644 index 0000000000..13eb5dda1a --- /dev/null +++ b/extensions/extention-sqs/pom.xml @@ -0,0 +1,80 @@ + + + + + extensions + org.thingsboard + 1.4.0-SNAPSHOT + + 4.0.0 + org.thingsboard.extensions + extension-sqs + jar + + Thingsboard Server SQS Extension + http://thingsboard.org + + + UTF-8 + ${basedir}/../.. + 1.11.229 + + + + + org.thingsboard + extensions-api + provided + + + com.amazonaws + aws-java-sdk-sqs + ${aws.sdk.version} + + + org.thingsboard + extensions-core + + + + + + + maven-assembly-plugin + + + src/assembly/extension.xml + + + + + make-assembly + package + + single + + + + + + + + \ No newline at end of file diff --git a/extensions/extention-sqs/src/assembly/extension.xml b/extensions/extention-sqs/src/assembly/extension.xml new file mode 100644 index 0000000000..2395d5551a --- /dev/null +++ b/extensions/extention-sqs/src/assembly/extension.xml @@ -0,0 +1,37 @@ + + + extension + + jar + + false + + + / + true + true + runtime + + + + + + \ No newline at end of file diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionMsg.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionMsg.java new file mode 100644 index 0000000000..c4650636f3 --- /dev/null +++ b/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 { + + public SqsFifoQueueActionMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, SqsFifoQueueActionPayload payload) { + super(tenantId, customerId, deviceId, payload); + } +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionPayload.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueueActionPayload.java new file mode 100644 index 0000000000..692cd90849 --- /dev/null +++ b/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; + +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java new file mode 100644 index 0000000000..f107ea103c --- /dev/null +++ b/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 { + + @Override + protected Optional 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())); + } +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginActionConfiguration.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginActionConfiguration.java new file mode 100644 index 0000000000..d10032576b --- /dev/null +++ b/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; +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionMsg.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionMsg.java new file mode 100644 index 0000000000..6666b274f3 --- /dev/null +++ b/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 { + + public SqsStandardQueueActionMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, SqsStandardQueueActionPayload payload) { + super(tenantId, customerId, deviceId, payload); + } +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionPayload.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueueActionPayload.java new file mode 100644 index 0000000000..ce3dd27a8d --- /dev/null +++ b/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; + +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java new file mode 100644 index 0000000000..1dd9d56d8a --- /dev/null +++ b/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 { + + @Override + protected Optional 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())); + } +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginActionConfiguration.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginActionConfiguration.java new file mode 100644 index 0000000000..5c21ef5840 --- /dev/null +++ b/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; + +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java new file mode 100644 index 0000000000..b71e0b18b7 --- /dev/null +++ b/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()))); + } + } +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPlugin.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPlugin.java new file mode 100644 index 0000000000..3f0252ebf6 --- /dev/null +++ b/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 { + + 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(); + } +} diff --git a/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPluginConfiguration.java b/extensions/extention-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsPluginConfiguration.java new file mode 100644 index 0000000000..a93b7f3f1d --- /dev/null +++ b/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; + +} diff --git a/extensions/extention-sqs/src/main/resources/SqsFifoQueueActionDescriptor.json b/extensions/extention-sqs/src/main/resources/SqsFifoQueueActionDescriptor.json new file mode 100644 index 0000000000..05bfaec11f --- /dev/null +++ b/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 + } + ] +} \ No newline at end of file diff --git a/extensions/extention-sqs/src/main/resources/SqsPluginDescriptor.json b/extensions/extention-sqs/src/main/resources/SqsPluginDescriptor.json new file mode 100644 index 0000000000..407468b5c8 --- /dev/null +++ b/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" + ] +} \ No newline at end of file diff --git a/extensions/extention-sqs/src/main/resources/SqsStandardQueueActionDescriptor.json b/extensions/extention-sqs/src/main/resources/SqsStandardQueueActionDescriptor.json new file mode 100644 index 0000000000..f5502e6a25 --- /dev/null +++ b/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 + } + ] +} \ No newline at end of file diff --git a/extensions/extention-sqs/src/test/java/org/thingsboard/server/extensions/sqs/SqsDemoClient.java b/extensions/extention-sqs/src/test/java/org/thingsboard/server/extensions/sqs/SqsDemoClient.java new file mode 100644 index 0000000000..17c9b08e8c --- /dev/null +++ b/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 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(); + } + } + } +} diff --git a/extensions/extention-sqs/src/test/resources/logback.xml b/extensions/extention-sqs/src/test/resources/logback.xml new file mode 100644 index 0000000000..757aa32dea --- /dev/null +++ b/extensions/extention-sqs/src/test/resources/logback.xml @@ -0,0 +1,10 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + \ No newline at end of file diff --git a/extensions/pom.xml b/extensions/pom.xml index 30c939556e..e1c20840ce 100644 --- a/extensions/pom.xml +++ b/extensions/pom.xml @@ -39,6 +39,8 @@ extension-rest-api-call extension-kafka extension-mqtt + extention-sqs + extension-sns diff --git a/pom.xml b/pom.xml index 1f9a92d97b..e45fc3ea28 100755 --- a/pom.xml +++ b/pom.xml @@ -350,6 +350,18 @@ extension ${project.version} + + org.thingsboard.extensions + extension-sqs + extension + ${project.version} + + + org.thingsboard.extensions + extension-sns + extension + ${project.version} + org.thingsboard.common data diff --git a/resume.bat b/resume.bat new file mode 100644 index 0000000000..c3c0e1d817 --- /dev/null +++ b/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 + diff --git a/tools/src/main/python/mqtt-send-telemetry.py b/tools/src/main/python/mqtt-send-telemetry.py new file mode 100644 index 0000000000..b9af45a0b1 --- /dev/null +++ b/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)