Andrew Shvayka
8 years ago
47 changed files with 1719 additions and 1067 deletions
@ -0,0 +1,88 @@ |
|||
<!-- |
|||
|
|||
Copyright © 2016-2018 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"> |
|||
<modelVersion>4.0.0</modelVersion> |
|||
<parent> |
|||
<groupId>org.thingsboard.common</groupId> |
|||
<version>2.2.0-SNAPSHOT</version> |
|||
<artifactId>transport</artifactId> |
|||
</parent> |
|||
<groupId>org.thingsboard.common.transport</groupId> |
|||
<artifactId>coap</artifactId> |
|||
<packaging>jar</packaging> |
|||
|
|||
<name>Thingsboard CoAP Transport Common</name> |
|||
<url>https://thingsboard.io</url> |
|||
|
|||
<properties> |
|||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> |
|||
<main.dir>${basedir}/../../..</main.dir> |
|||
</properties> |
|||
|
|||
<dependencies> |
|||
<dependency> |
|||
<groupId>org.thingsboard.common.transport</groupId> |
|||
<artifactId>transport-api</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.eclipse.californium</groupId> |
|||
<artifactId>californium-core</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.springframework</groupId> |
|||
<artifactId>spring-context-support</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.springframework</groupId> |
|||
<artifactId>spring-context</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.slf4j</groupId> |
|||
<artifactId>slf4j-api</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.slf4j</groupId> |
|||
<artifactId>log4j-over-slf4j</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>ch.qos.logback</groupId> |
|||
<artifactId>logback-core</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>ch.qos.logback</groupId> |
|||
<artifactId>logback-classic</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.springframework.boot</groupId> |
|||
<artifactId>spring-boot-starter-test</artifactId> |
|||
<scope>test</scope> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>junit</groupId> |
|||
<artifactId>junit</artifactId> |
|||
<scope>test</scope> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.mockito</groupId> |
|||
<artifactId>mockito-all</artifactId> |
|||
<scope>test</scope> |
|||
</dependency> |
|||
</dependencies> |
|||
|
|||
</project> |
|||
@ -0,0 +1,51 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.transport.TransportContext; |
|||
import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor; |
|||
|
|||
/** |
|||
* Created by ashvayka on 18.10.18. |
|||
*/ |
|||
@Slf4j |
|||
@ConditionalOnExpression("'${transport.type:null}'=='null' || ('${transport.type}'=='local' && '${transport.coap.enabled}'=='true')") |
|||
@Component |
|||
public class CoapTransportContext extends TransportContext { |
|||
|
|||
@Getter |
|||
@Value("${transport.coap.bind_address}") |
|||
private String host; |
|||
|
|||
@Getter |
|||
@Value("${transport.coap.bind_port}") |
|||
private Integer port; |
|||
|
|||
@Getter |
|||
@Value("${transport.coap.timeout}") |
|||
private Long timeout; |
|||
|
|||
@Getter |
|||
@Autowired |
|||
private CoapTransportAdaptor adaptor; |
|||
|
|||
} |
|||
@ -0,0 +1,445 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.californium.core.CoapResource; |
|||
import org.eclipse.californium.core.coap.CoAP.ResponseCode; |
|||
import org.eclipse.californium.core.coap.Request; |
|||
import org.eclipse.californium.core.coap.Response; |
|||
import org.eclipse.californium.core.network.Exchange; |
|||
import org.eclipse.californium.core.network.ExchangeObserver; |
|||
import org.eclipse.californium.core.server.resources.CoapExchange; |
|||
import org.eclipse.californium.core.server.resources.Resource; |
|||
import org.springframework.util.ReflectionUtils; |
|||
import org.thingsboard.server.common.data.security.DeviceTokenCredentials; |
|||
import org.thingsboard.server.common.msg.session.FeatureType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.transport.SessionMsgListener; |
|||
import org.thingsboard.server.common.transport.TransportContext; |
|||
import org.thingsboard.server.common.transport.TransportService; |
|||
import org.thingsboard.server.common.transport.TransportServiceCallback; |
|||
import org.thingsboard.server.common.transport.adaptor.AdaptorException; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
|
|||
import java.lang.reflect.Field; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.function.Consumer; |
|||
|
|||
@Slf4j |
|||
public class CoapTransportResource extends CoapResource { |
|||
// coap://localhost:port/api/v1/DEVICE_TOKEN/[attributes|telemetry|rpc[/requestId]]
|
|||
private static final int ACCESS_TOKEN_POSITION = 3; |
|||
private static final int FEATURE_TYPE_POSITION = 4; |
|||
private static final int REQUEST_ID_POSITION = 5; |
|||
|
|||
private final CoapTransportContext transportContext; |
|||
private final TransportService transportService; |
|||
private final Field observerField; |
|||
private final long timeout; |
|||
private final ConcurrentMap<String, TransportProtos.SessionInfoProto> tokenToSessionIdMap = new ConcurrentHashMap<>(); |
|||
|
|||
public CoapTransportResource(CoapTransportContext context, String name) { |
|||
super(name); |
|||
this.transportContext = context; |
|||
this.transportService = context.getTransportService(); |
|||
this.timeout = context.getTimeout(); |
|||
// This is important to turn off existing observable logic in
|
|||
// CoapResource. We will have our own observe monitoring due to 1:1
|
|||
// observe relationship.
|
|||
this.setObservable(false); |
|||
observerField = ReflectionUtils.findField(Exchange.class, "observer"); |
|||
observerField.setAccessible(true); |
|||
} |
|||
|
|||
@Override |
|||
public void handleGET(CoapExchange exchange) { |
|||
if (transportContext.getQuotaService().isQuotaExceeded(exchange.getSourceAddress().getHostAddress())) { |
|||
log.warn("CoAP Quota exceeded for [{}:{}] . Disconnect", exchange.getSourceAddress().getHostAddress(), exchange.getSourcePort()); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
return; |
|||
} |
|||
|
|||
Optional<FeatureType> featureType = getFeatureType(exchange.advanced().getRequest()); |
|||
if (!featureType.isPresent()) { |
|||
log.trace("Missing feature type parameter"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} else if (featureType.get() == FeatureType.TELEMETRY) { |
|||
log.trace("Can't fetch/subscribe to timeseries updates"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} else if (exchange.getRequestOptions().hasObserve()) { |
|||
processExchangeGetRequest(exchange, featureType.get()); |
|||
} else if (featureType.get() == FeatureType.ATTRIBUTES) { |
|||
processRequest(exchange, SessionMsgType.GET_ATTRIBUTES_REQUEST); |
|||
} else { |
|||
log.trace("Invalid feature type parameter"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} |
|||
} |
|||
|
|||
private void processExchangeGetRequest(CoapExchange exchange, FeatureType featureType) { |
|||
boolean unsubscribe = exchange.getRequestOptions().getObserve() == 1; |
|||
SessionMsgType sessionMsgType; |
|||
if (featureType == FeatureType.RPC) { |
|||
sessionMsgType = unsubscribe ? SessionMsgType.UNSUBSCRIBE_RPC_COMMANDS_REQUEST : SessionMsgType.SUBSCRIBE_RPC_COMMANDS_REQUEST; |
|||
} else { |
|||
sessionMsgType = unsubscribe ? SessionMsgType.UNSUBSCRIBE_ATTRIBUTES_REQUEST : SessionMsgType.SUBSCRIBE_ATTRIBUTES_REQUEST; |
|||
} |
|||
processRequest(exchange, sessionMsgType); |
|||
} |
|||
|
|||
@Override |
|||
public void handlePOST(CoapExchange exchange) { |
|||
if (transportContext.getQuotaService().isQuotaExceeded(exchange.getSourceAddress().getHostAddress())) { |
|||
log.warn("CoAP Quota exceeded for [{}:{}] . Disconnect", exchange.getSourceAddress().getHostAddress(), exchange.getSourcePort()); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
return; |
|||
} |
|||
|
|||
Optional<FeatureType> featureType = getFeatureType(exchange.advanced().getRequest()); |
|||
if (!featureType.isPresent()) { |
|||
log.trace("Missing feature type parameter"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} else { |
|||
switch (featureType.get()) { |
|||
case ATTRIBUTES: |
|||
processRequest(exchange, SessionMsgType.POST_ATTRIBUTES_REQUEST); |
|||
break; |
|||
case TELEMETRY: |
|||
processRequest(exchange, SessionMsgType.POST_TELEMETRY_REQUEST); |
|||
break; |
|||
case RPC: |
|||
Optional<Integer> requestId = getRequestId(exchange.advanced().getRequest()); |
|||
if (requestId.isPresent()) { |
|||
processRequest(exchange, SessionMsgType.TO_DEVICE_RPC_RESPONSE); |
|||
} else { |
|||
processRequest(exchange, SessionMsgType.TO_SERVER_RPC_REQUEST); |
|||
} |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void processRequest(CoapExchange exchange, SessionMsgType type) { |
|||
log.trace("Processing {}", exchange.advanced().getRequest()); |
|||
exchange.accept(); |
|||
Exchange advanced = exchange.advanced(); |
|||
Request request = advanced.getRequest(); |
|||
|
|||
Optional<DeviceTokenCredentials> credentials = decodeCredentials(request); |
|||
if (!credentials.isPresent()) { |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
return; |
|||
} |
|||
|
|||
transportService.process(TransportProtos.ValidateDeviceTokenRequestMsg.newBuilder().setToken(credentials.get().getCredentialsId()).build(), |
|||
new DeviceAuthCallback(transportContext, exchange, sessionInfo -> { |
|||
UUID sessionId = new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()); |
|||
try { |
|||
switch (type) { |
|||
case POST_ATTRIBUTES_REQUEST: |
|||
transportService.process(sessionInfo, |
|||
transportContext.getAdaptor().convertToPostAttributes(sessionId, request), |
|||
new CoapOkCallback(exchange)); |
|||
break; |
|||
case POST_TELEMETRY_REQUEST: |
|||
transportService.process(sessionInfo, |
|||
transportContext.getAdaptor().convertToPostTelemetry(sessionId, request), |
|||
new CoapOkCallback(exchange)); |
|||
break; |
|||
case SUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
advanced.setObserver(new CoapExchangeObserverProxy((ExchangeObserver) observerField.get(advanced), |
|||
registerAsyncCoapSession(exchange, request, sessionInfo, sessionId))); |
|||
transportService.process(sessionInfo, |
|||
TransportProtos.SubscribeToAttributeUpdatesMsg.getDefaultInstance(), |
|||
new CoapNoOpCallback(exchange)); |
|||
break; |
|||
case UNSUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
TransportProtos.SessionInfoProto attrSession = lookupAsyncSessionInfo(request); |
|||
if (attrSession != null) { |
|||
transportService.process(attrSession, |
|||
TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), |
|||
new CoapOkCallback(exchange)); |
|||
closeAndDeregister(sessionInfo); |
|||
} |
|||
break; |
|||
case SUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
advanced.setObserver(new CoapExchangeObserverProxy((ExchangeObserver) observerField.get(advanced), |
|||
registerAsyncCoapSession(exchange, request, sessionInfo, sessionId))); |
|||
transportService.process(sessionInfo, |
|||
TransportProtos.SubscribeToRPCMsg.getDefaultInstance(), |
|||
new CoapNoOpCallback(exchange)); |
|||
break; |
|||
case UNSUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
TransportProtos.SessionInfoProto rpcSession = lookupAsyncSessionInfo(request); |
|||
if (rpcSession != null) { |
|||
transportService.process(rpcSession, |
|||
TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), |
|||
new CoapOkCallback(exchange)); |
|||
transportService.process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); |
|||
transportService.deregisterSession(rpcSession); |
|||
} |
|||
break; |
|||
case TO_DEVICE_RPC_RESPONSE: |
|||
transportService.process(sessionInfo, |
|||
transportContext.getAdaptor().convertToDeviceRpcResponse(sessionId, request), |
|||
new CoapOkCallback(exchange)); |
|||
break; |
|||
case TO_SERVER_RPC_REQUEST: |
|||
transportService.process(sessionInfo, |
|||
transportContext.getAdaptor().convertToServerRpcRequest(sessionId, request), |
|||
new CoapNoOpCallback(exchange)); |
|||
break; |
|||
case GET_ATTRIBUTES_REQUEST: |
|||
transportService.registerSyncSession(sessionInfo, new CoapSessionListener(sessionId, exchange), transportContext.getTimeout()); |
|||
transportService.process(sessionInfo, |
|||
transportContext.getAdaptor().convertToGetAttributes(sessionId, request), |
|||
new CoapNoOpCallback(exchange)); |
|||
break; |
|||
} |
|||
} catch (AdaptorException e) { |
|||
log.trace("[{}] Failed to decode message: ", sessionId, e); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} catch (IllegalAccessException e) { |
|||
log.trace("[{}] Failed to process message: ", sessionId, e); |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
})); |
|||
} |
|||
|
|||
private TransportProtos.SessionInfoProto lookupAsyncSessionInfo(Request request) { |
|||
String token = request.getSource().getHostAddress() + ":" + request.getSourcePort() + ":" + request.getTokenString(); |
|||
return tokenToSessionIdMap.remove(token); |
|||
} |
|||
|
|||
private String registerAsyncCoapSession(CoapExchange exchange, Request request, TransportProtos.SessionInfoProto sessionInfo, UUID sessionId) { |
|||
String token = request.getSource().getHostAddress() + ":" + request.getSourcePort() + ":" + request.getTokenString(); |
|||
tokenToSessionIdMap.putIfAbsent(token, sessionInfo); |
|||
CoapSessionListener attrListener = new CoapSessionListener(sessionId, exchange); |
|||
transportService.registerAsyncSession(sessionInfo, attrListener); |
|||
transportService.process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null); |
|||
return token; |
|||
} |
|||
|
|||
private static TransportProtos.SessionEventMsg getSessionEventMsg(TransportProtos.SessionEvent event) { |
|||
return TransportProtos.SessionEventMsg.newBuilder() |
|||
.setSessionType(TransportProtos.SessionType.ASYNC) |
|||
.setEvent(event).build(); |
|||
} |
|||
|
|||
private Optional<DeviceTokenCredentials> decodeCredentials(Request request) { |
|||
List<String> uriPath = request.getOptions().getUriPath(); |
|||
if (uriPath.size() >= ACCESS_TOKEN_POSITION) { |
|||
return Optional.of(new DeviceTokenCredentials(uriPath.get(ACCESS_TOKEN_POSITION - 1))); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
} |
|||
|
|||
private Optional<FeatureType> getFeatureType(Request request) { |
|||
List<String> uriPath = request.getOptions().getUriPath(); |
|||
try { |
|||
if (uriPath.size() >= FEATURE_TYPE_POSITION) { |
|||
return Optional.of(FeatureType.valueOf(uriPath.get(FEATURE_TYPE_POSITION - 1).toUpperCase())); |
|||
} |
|||
} catch (RuntimeException e) { |
|||
log.warn("Failed to decode feature type: {}", uriPath); |
|||
} |
|||
return Optional.empty(); |
|||
} |
|||
|
|||
public static Optional<Integer> getRequestId(Request request) { |
|||
List<String> uriPath = request.getOptions().getUriPath(); |
|||
try { |
|||
if (uriPath.size() >= REQUEST_ID_POSITION) { |
|||
return Optional.of(Integer.valueOf(uriPath.get(REQUEST_ID_POSITION - 1))); |
|||
} |
|||
} catch (RuntimeException e) { |
|||
log.warn("Failed to decode feature type: {}", uriPath); |
|||
} |
|||
return Optional.empty(); |
|||
} |
|||
|
|||
@Override |
|||
public Resource getChild(String name) { |
|||
return this; |
|||
} |
|||
|
|||
private static class DeviceAuthCallback implements TransportServiceCallback<TransportProtos.ValidateDeviceCredentialsResponseMsg> { |
|||
private final TransportContext transportContext; |
|||
private final CoapExchange exchange; |
|||
private final Consumer<TransportProtos.SessionInfoProto> onSuccess; |
|||
|
|||
DeviceAuthCallback(TransportContext transportContext, CoapExchange exchange, Consumer<TransportProtos.SessionInfoProto> onSuccess) { |
|||
this.transportContext = transportContext; |
|||
this.exchange = exchange; |
|||
this.onSuccess = onSuccess; |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess(TransportProtos.ValidateDeviceCredentialsResponseMsg msg) { |
|||
if (msg.hasDeviceInfo()) { |
|||
UUID sessionId = UUID.randomUUID(); |
|||
TransportProtos.DeviceInfoProto deviceInfoProto = msg.getDeviceInfo(); |
|||
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
|||
.setNodeId(transportContext.getNodeId()) |
|||
.setTenantIdMSB(deviceInfoProto.getTenantIdMSB()) |
|||
.setTenantIdLSB(deviceInfoProto.getTenantIdLSB()) |
|||
.setDeviceIdMSB(deviceInfoProto.getDeviceIdMSB()) |
|||
.setDeviceIdLSB(deviceInfoProto.getDeviceIdLSB()) |
|||
.setSessionIdMSB(sessionId.getMostSignificantBits()) |
|||
.setSessionIdLSB(sessionId.getLeastSignificantBits()) |
|||
.build(); |
|||
onSuccess.accept(sessionInfo); |
|||
} else { |
|||
exchange.respond(ResponseCode.UNAUTHORIZED); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onError(Throwable e) { |
|||
log.warn("Failed to process request", e); |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
private static class CoapOkCallback implements TransportServiceCallback<Void> { |
|||
private final CoapExchange exchange; |
|||
|
|||
CoapOkCallback(CoapExchange exchange) { |
|||
this.exchange = exchange; |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess(Void msg) { |
|||
exchange.respond(ResponseCode.VALID); |
|||
} |
|||
|
|||
@Override |
|||
public void onError(Throwable e) { |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
private static class CoapNoOpCallback implements TransportServiceCallback<Void> { |
|||
private final CoapExchange exchange; |
|||
|
|||
CoapNoOpCallback(CoapExchange exchange) { |
|||
this.exchange = exchange; |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess(Void msg) { |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public void onError(Throwable e) { |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
public class CoapSessionListener implements SessionMsgListener { |
|||
|
|||
private final CoapExchange exchange; |
|||
private final AtomicInteger seqNumber = new AtomicInteger(2); |
|||
|
|||
CoapSessionListener(UUID sessionId, CoapExchange exchange) { |
|||
this.exchange = exchange; |
|||
} |
|||
|
|||
@Override |
|||
public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg msg) { |
|||
try { |
|||
exchange.respond(transportContext.getAdaptor().convertToPublish(this, msg)); |
|||
} catch (AdaptorException e) { |
|||
log.trace("Failed to reply due to error", e); |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onAttributeUpdate(TransportProtos.AttributeUpdateNotificationMsg msg) { |
|||
try { |
|||
exchange.respond(transportContext.getAdaptor().convertToPublish(this, msg)); |
|||
} catch (AdaptorException e) { |
|||
log.trace("Failed to reply due to error", e); |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto sessionCloseNotification) { |
|||
exchange.respond(ResponseCode.SERVICE_UNAVAILABLE); |
|||
} |
|||
|
|||
@Override |
|||
public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg msg) { |
|||
try { |
|||
exchange.respond(transportContext.getAdaptor().convertToPublish(this, msg)); |
|||
} catch (AdaptorException e) { |
|||
log.trace("Failed to reply due to error", e); |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg msg) { |
|||
try { |
|||
exchange.respond(transportContext.getAdaptor().convertToPublish(this, msg)); |
|||
} catch (AdaptorException e) { |
|||
log.trace("Failed to reply due to error", e); |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
public int getNextSeqNumber() { |
|||
return seqNumber.getAndIncrement(); |
|||
} |
|||
} |
|||
|
|||
public class CoapExchangeObserverProxy implements ExchangeObserver { |
|||
|
|||
private final ExchangeObserver proxy; |
|||
private final String token; |
|||
|
|||
CoapExchangeObserverProxy(ExchangeObserver proxy, String token) { |
|||
super(); |
|||
this.proxy = proxy; |
|||
this.token = token; |
|||
} |
|||
|
|||
@Override |
|||
public void completed(Exchange exchange) { |
|||
proxy.completed(exchange); |
|||
TransportProtos.SessionInfoProto session = tokenToSessionIdMap.remove(token); |
|||
if (session != null) { |
|||
closeAndDeregister(session); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void closeAndDeregister(TransportProtos.SessionInfoProto session) { |
|||
transportService.process(session, getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); |
|||
transportService.deregisterSession(session); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap.adaptors; |
|||
|
|||
import org.eclipse.californium.core.coap.Request; |
|||
import org.eclipse.californium.core.coap.Response; |
|||
import org.thingsboard.server.common.transport.adaptor.AdaptorException; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.transport.coap.CoapTransportResource; |
|||
|
|||
import java.util.UUID; |
|||
import java.util.Optional; |
|||
|
|||
public interface CoapTransportAdaptor { |
|||
|
|||
TransportProtos.PostTelemetryMsg convertToPostTelemetry(UUID sessionId, Request inbound) throws AdaptorException; |
|||
|
|||
TransportProtos.PostAttributeMsg convertToPostAttributes(UUID sessionId, Request inbound) throws AdaptorException; |
|||
|
|||
TransportProtos.GetAttributeRequestMsg convertToGetAttributes(UUID sessionId, Request inbound) throws AdaptorException; |
|||
|
|||
TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound) throws AdaptorException; |
|||
|
|||
TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(UUID sessionId, Request inbound) throws AdaptorException; |
|||
|
|||
Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException; |
|||
|
|||
Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException; |
|||
|
|||
Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException; |
|||
|
|||
Response convertToPublish(CoapTransportResource.CoapSessionListener coapSessionListener, TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException; |
|||
|
|||
} |
|||
@ -0,0 +1,155 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap.adaptors; |
|||
|
|||
import java.util.*; |
|||
|
|||
import com.google.gson.JsonElement; |
|||
import com.google.gson.JsonObject; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.californium.core.coap.CoAP; |
|||
import org.eclipse.californium.core.coap.Request; |
|||
import org.eclipse.californium.core.coap.Response; |
|||
import org.springframework.util.StringUtils; |
|||
import org.thingsboard.server.common.msg.kv.AttributesKVMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionContext; |
|||
import org.thingsboard.server.common.transport.adaptor.AdaptorException; |
|||
import org.thingsboard.server.common.transport.adaptor.JsonConverter; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import com.google.gson.JsonParser; |
|||
import com.google.gson.JsonSyntaxException; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.transport.coap.CoapTransportResource; |
|||
|
|||
@Component("JsonCoapAdaptor") |
|||
@Slf4j |
|||
public class JsonCoapAdaptor implements CoapTransportAdaptor { |
|||
|
|||
@Override |
|||
public TransportProtos.PostTelemetryMsg convertToPostTelemetry(UUID sessionId, Request inbound) throws AdaptorException { |
|||
String payload = validatePayload(sessionId, inbound); |
|||
try { |
|||
return JsonConverter.convertToTelemetryProto(new JsonParser().parse(payload)); |
|||
} catch (IllegalStateException | JsonSyntaxException ex) { |
|||
throw new AdaptorException(ex); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public TransportProtos.PostAttributeMsg convertToPostAttributes(UUID sessionId, Request inbound) throws AdaptorException { |
|||
String payload = validatePayload(sessionId, inbound); |
|||
try { |
|||
return JsonConverter.convertToAttributesProto(new JsonParser().parse(payload)); |
|||
} catch (IllegalStateException | JsonSyntaxException ex) { |
|||
throw new AdaptorException(ex); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(UUID sessionId, Request inbound) throws AdaptorException { |
|||
List<String> queryElements = inbound.getOptions().getUriQuery(); |
|||
TransportProtos.GetAttributeRequestMsg.Builder result = TransportProtos.GetAttributeRequestMsg.newBuilder(); |
|||
if (queryElements != null && queryElements.size() > 0) { |
|||
Set<String> clientKeys = toKeys(queryElements, "clientKeys"); |
|||
Set<String> sharedKeys = toKeys(queryElements, "sharedKeys"); |
|||
if (clientKeys != null) { |
|||
result.addAllClientAttributeNames(clientKeys); |
|||
} |
|||
if (sharedKeys != null) { |
|||
result.addAllSharedAttributeNames(sharedKeys); |
|||
} |
|||
} |
|||
return result.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound) throws AdaptorException { |
|||
Optional<Integer> requestId = CoapTransportResource.getRequestId(inbound); |
|||
String payload = validatePayload(sessionId, inbound); |
|||
JsonObject response = new JsonParser().parse(payload).getAsJsonObject(); |
|||
return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId.orElseThrow(() -> new AdaptorException("Request id is missing!"))) |
|||
.setPayload(response.toString()).build(); |
|||
} |
|||
|
|||
@Override |
|||
public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(UUID sessionId, Request inbound) throws AdaptorException { |
|||
String payload = validatePayload(sessionId, inbound); |
|||
return JsonConverter.convertToServerRpcRequest(new JsonParser().parse(payload), 0); |
|||
} |
|||
|
|||
@Override |
|||
public Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.AttributeUpdateNotificationMsg msg) throws AdaptorException { |
|||
return getObserveNotification(session.getNextSeqNumber(), JsonConverter.toJson(msg)); |
|||
} |
|||
|
|||
@Override |
|||
public Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.ToDeviceRpcRequestMsg msg) throws AdaptorException { |
|||
return getObserveNotification(session.getNextSeqNumber(), JsonConverter.toJson(msg, true)); |
|||
} |
|||
|
|||
@Override |
|||
public Response convertToPublish(CoapTransportResource.CoapSessionListener coapSessionListener, TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException { |
|||
Response response = new Response(CoAP.ResponseCode.CONTENT); |
|||
JsonElement result = JsonConverter.toJson(msg); |
|||
response.setPayload(result.toString()); |
|||
return response; |
|||
} |
|||
|
|||
@Override |
|||
public Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.GetAttributeResponseMsg msg) throws AdaptorException { |
|||
if (msg.getClientAttributeListCount() == 0 && msg.getSharedAttributeListCount() == 0 && msg.getDeletedAttributeKeysCount() == 0) { |
|||
return new Response(CoAP.ResponseCode.NOT_FOUND); |
|||
} else { |
|||
Response response = new Response(CoAP.ResponseCode.CONTENT); |
|||
JsonObject result = JsonConverter.toJson(msg); |
|||
response.setPayload(result.toString()); |
|||
return response; |
|||
} |
|||
} |
|||
|
|||
private Response getObserveNotification(int seqNumber, JsonElement json) { |
|||
Response response = new Response(CoAP.ResponseCode.CONTENT); |
|||
response.getOptions().setObserve(seqNumber); |
|||
response.setPayload(json.toString()); |
|||
return response; |
|||
} |
|||
|
|||
private String validatePayload(UUID sessionId, Request inbound) throws AdaptorException { |
|||
String payload = inbound.getPayloadString(); |
|||
if (payload == null) { |
|||
log.warn("[{}] Payload is empty!", sessionId); |
|||
throw new AdaptorException(new IllegalArgumentException("Payload is empty!")); |
|||
} |
|||
return payload; |
|||
} |
|||
|
|||
private Set<String> toKeys(List<String> queryElements, String attributeName) throws AdaptorException { |
|||
String keys = null; |
|||
for (String queryElement : queryElements) { |
|||
String[] queryItem = queryElement.split("="); |
|||
if (queryItem.length == 2 && queryItem[0].equals(attributeName)) { |
|||
keys = queryItem[1]; |
|||
} |
|||
} |
|||
if (keys != null && !StringUtils.isEmpty(keys)) { |
|||
return new HashSet<>(Arrays.asList(keys.split(","))); |
|||
} else { |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,140 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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. |
|||
*/ |
|||
import org.apache.tools.ant.filters.ReplaceTokens |
|||
|
|||
buildscript { |
|||
ext { |
|||
osPackageVersion = "3.8.0" |
|||
} |
|||
repositories { |
|||
jcenter() |
|||
} |
|||
dependencies { |
|||
classpath("com.netflix.nebula:gradle-ospackage-plugin:${osPackageVersion}") |
|||
} |
|||
} |
|||
|
|||
apply plugin: "nebula.ospackage" |
|||
|
|||
buildDir = projectBuildDir |
|||
version = projectVersion |
|||
distsDirName = "./" |
|||
|
|||
// OS Package plugin configuration |
|||
ospackage { |
|||
packageName = pkgName |
|||
version = "${project.version}" |
|||
release = 1 |
|||
os = LINUX |
|||
type = BINARY |
|||
|
|||
into pkgInstallFolder |
|||
|
|||
user pkgName |
|||
permissionGroup pkgName |
|||
|
|||
// Copy the actual .jar file |
|||
from(mainJar) { |
|||
// Strip the version from the jar filename |
|||
rename { String fileName -> |
|||
"${pkgName}.jar" |
|||
} |
|||
fileMode 0500 |
|||
into "bin" |
|||
} |
|||
|
|||
// Copy the config files |
|||
from("target/conf") { |
|||
exclude "${pkgName}.conf" |
|||
fileType CONFIG | NOREPLACE |
|||
fileMode 0754 |
|||
into "conf" |
|||
} |
|||
|
|||
} |
|||
|
|||
// Configure our RPM build task |
|||
buildRpm { |
|||
|
|||
arch = NOARCH |
|||
|
|||
version = projectVersion.replace('-', '') |
|||
archiveName = "${pkgName}.rpm" |
|||
|
|||
requires("java-1.8.0") |
|||
|
|||
from("target/conf") { |
|||
include "${pkgName}.conf" |
|||
filter(ReplaceTokens, tokens: ['pkg.platform': 'rpm']) |
|||
fileType CONFIG | NOREPLACE |
|||
fileMode 0754 |
|||
into "${pkgInstallFolder}/conf" |
|||
} |
|||
|
|||
preInstall file("${buildDir}/control/rpm/preinst") |
|||
postInstall file("${buildDir}/control/rpm/postinst") |
|||
preUninstall file("${buildDir}/control/rpm/prerm") |
|||
postUninstall file("${buildDir}/control/rpm/postrm") |
|||
|
|||
user pkgName |
|||
permissionGroup pkgName |
|||
|
|||
// Copy the system unit files |
|||
from("${buildDir}/control/${pkgName}.service") { |
|||
addParentDirs = false |
|||
fileMode 0644 |
|||
into "/usr/lib/systemd/system" |
|||
} |
|||
|
|||
directory(pkgLogFolder, 0755) |
|||
link("${pkgInstallFolder}/bin/${pkgName}.yml", "${pkgInstallFolder}/conf/${pkgName}.yml") |
|||
link("/etc/${pkgName}/conf", "${pkgInstallFolder}/conf") |
|||
} |
|||
|
|||
// Same as the buildRpm task |
|||
buildDeb { |
|||
|
|||
arch = "all" |
|||
|
|||
archiveName = "${pkgName}.deb" |
|||
|
|||
requires("openjdk-8-jre").or("java8-runtime").or("oracle-java8-installer").or("openjdk-8-jre-headless") |
|||
|
|||
from("target/conf") { |
|||
include "${pkgName}.conf" |
|||
filter(ReplaceTokens, tokens: ['pkg.platform': 'deb']) |
|||
fileType CONFIG | NOREPLACE |
|||
fileMode 0754 |
|||
into "${pkgInstallFolder}/conf" |
|||
} |
|||
|
|||
configurationFile("${pkgInstallFolder}/conf/${pkgName}.conf") |
|||
configurationFile("${pkgInstallFolder}/conf/${pkgName}.yml") |
|||
configurationFile("${pkgInstallFolder}/conf/logback.xml") |
|||
|
|||
preInstall file("${buildDir}/control/deb/preinst") |
|||
postInstall file("${buildDir}/control/deb/postinst") |
|||
preUninstall file("${buildDir}/control/deb/prerm") |
|||
postUninstall file("${buildDir}/control/deb/postrm") |
|||
|
|||
user pkgName |
|||
permissionGroup pkgName |
|||
|
|||
directory(pkgLogFolder, 0755) |
|||
link("/etc/init.d/${pkgName}", "${pkgInstallFolder}/bin/${pkgName}.jar") |
|||
link("${pkgInstallFolder}/bin/${pkgName}.yml", "${pkgInstallFolder}/conf/${pkgName}.yml") |
|||
link("/etc/${pkgName}/conf", "${pkgInstallFolder}/conf") |
|||
} |
|||
@ -0,0 +1,71 @@ |
|||
<!-- |
|||
|
|||
Copyright © 2016-2018 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/plugins/maven-assembly-plugin/assembly/1.1.3" |
|||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" |
|||
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.3 http://maven.apache.org/xsd/assembly-1.1.3.xsd"> |
|||
<id>windows</id> |
|||
|
|||
<formats> |
|||
<format>zip</format> |
|||
</formats> |
|||
|
|||
<!-- Workaround to create logs directory --> |
|||
<fileSets> |
|||
<fileSet> |
|||
<directory>${pkg.win.dist}</directory> |
|||
<outputDirectory>logs</outputDirectory> |
|||
<excludes> |
|||
<exclude>*/**</exclude> |
|||
</excludes> |
|||
</fileSet> |
|||
<fileSet> |
|||
<directory>${pkg.win.dist}/conf</directory> |
|||
<outputDirectory>conf</outputDirectory> |
|||
<lineEnding>windows</lineEnding> |
|||
</fileSet> |
|||
</fileSets> |
|||
|
|||
<files> |
|||
<file> |
|||
<source>${project.build.directory}/${project.build.finalName}-boot.${project.packaging}</source> |
|||
<outputDirectory>lib</outputDirectory> |
|||
<destName>${pkg.name}.jar</destName> |
|||
</file> |
|||
<file> |
|||
<source>${pkg.win.dist}/service.exe</source> |
|||
<outputDirectory/> |
|||
<destName>${pkg.name}.exe</destName> |
|||
</file> |
|||
<file> |
|||
<source>${pkg.win.dist}/service.xml</source> |
|||
<outputDirectory/> |
|||
<destName>${pkg.name}.xml</destName> |
|||
<lineEnding>windows</lineEnding> |
|||
</file> |
|||
<file> |
|||
<source>${pkg.win.dist}/install.bat</source> |
|||
<outputDirectory/> |
|||
<lineEnding>windows</lineEnding> |
|||
</file> |
|||
<file> |
|||
<source>${pkg.win.dist}/uninstall.bat</source> |
|||
<outputDirectory/> |
|||
<lineEnding>windows</lineEnding> |
|||
</file> |
|||
</files> |
|||
</assembly> |
|||
@ -0,0 +1,43 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!-- |
|||
|
|||
Copyright © 2016-2018 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. |
|||
|
|||
--> |
|||
<!DOCTYPE configuration> |
|||
<configuration> |
|||
|
|||
<appender name="fileLogAppender" |
|||
class="ch.qos.logback.core.rolling.RollingFileAppender"> |
|||
<file>${pkg.logFolder}/${pkg.name}.log</file> |
|||
<rollingPolicy |
|||
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy"> |
|||
<fileNamePattern>${pkg.logFolder}/${pkg.name}.%d{yyyy-MM-dd}.%i.log</fileNamePattern> |
|||
<maxFileSize>100MB</maxFileSize> |
|||
<maxHistory>30</maxHistory> |
|||
<totalSizeCap>3GB</totalSizeCap> |
|||
</rollingPolicy> |
|||
<encoder> |
|||
<pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern> |
|||
</encoder> |
|||
</appender> |
|||
|
|||
<logger name="org.thingsboard.server" level="INFO" /> |
|||
|
|||
<root level="INFO"> |
|||
<appender-ref ref="fileLogAppender"/> |
|||
</root> |
|||
|
|||
</configuration> |
|||
@ -0,0 +1,23 @@ |
|||
# |
|||
# Copyright © 2016-2018 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. |
|||
# |
|||
|
|||
export JAVA_OPTS="$JAVA_OPTS -Xloggc:@pkg.logFolder@/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:+PrintGCDetails -XX:+PrintGCDateStamps" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:+CMSEdenChunksRecordAlways -XX:CMSInitiatingOccupancyFraction=75 -XX:+UseCMSInitiatingOccupancyOnly" |
|||
export LOG_FILENAME=${pkg.name}.out |
|||
export LOADER_PATH=${pkg.installFolder}/conf |
|||
@ -0,0 +1 @@ |
|||
pkg.logFolder=${pkg.unixLogFolder} |
|||
@ -0,0 +1,2 @@ |
|||
pkg.logFolder=${BASE}\\logs |
|||
pkg.winWrapperLogFolder=%BASE%\\logs |
|||
@ -0,0 +1,48 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.coap; |
|||
|
|||
import org.springframework.boot.SpringApplication; |
|||
import org.springframework.boot.SpringBootConfiguration; |
|||
import org.springframework.context.annotation.ComponentScan; |
|||
import org.springframework.scheduling.annotation.EnableAsync; |
|||
import org.springframework.scheduling.annotation.EnableScheduling; |
|||
|
|||
import java.util.Arrays; |
|||
|
|||
@SpringBootConfiguration |
|||
@EnableAsync |
|||
@EnableScheduling |
|||
@ComponentScan({"org.thingsboard.server.coap", "org.thingsboard.server.common", "org.thingsboard.server.transport.coap", "org.thingsboard.server.kafka"}) |
|||
public class ThingsboardCoapTransportApplication { |
|||
|
|||
private static final String SPRING_CONFIG_NAME_KEY = "--spring.config.name"; |
|||
private static final String DEFAULT_SPRING_CONFIG_PARAM = SPRING_CONFIG_NAME_KEY + "=" + "tb-coap-transport"; |
|||
|
|||
public static void main(String[] args) { |
|||
SpringApplication.run(ThingsboardCoapTransportApplication.class, updateArguments(args)); |
|||
} |
|||
|
|||
private static String[] updateArguments(String[] args) { |
|||
if (Arrays.stream(args).noneMatch(arg -> arg.startsWith(SPRING_CONFIG_NAME_KEY))) { |
|||
String[] modifiedArgs = new String[args.length + 1]; |
|||
System.arraycopy(args, 0, modifiedArgs, 0, args.length); |
|||
modifiedArgs[args.length] = DEFAULT_SPRING_CONFIG_PARAM; |
|||
return modifiedArgs; |
|||
} |
|||
return args; |
|||
} |
|||
} |
|||
@ -1,242 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap; |
|||
|
|||
import java.lang.reflect.Field; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.californium.core.CoapResource; |
|||
import org.eclipse.californium.core.coap.CoAP.ResponseCode; |
|||
import org.eclipse.californium.core.coap.Request; |
|||
import org.eclipse.californium.core.network.Exchange; |
|||
import org.eclipse.californium.core.network.ExchangeObserver; |
|||
import org.eclipse.californium.core.server.resources.CoapExchange; |
|||
import org.eclipse.californium.core.server.resources.Resource; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentialsFilter; |
|||
import org.thingsboard.server.common.data.security.DeviceTokenCredentials; |
|||
import org.thingsboard.server.common.msg.session.*; |
|||
import org.thingsboard.server.common.transport.SessionMsgProcessor; |
|||
import org.thingsboard.server.common.transport.adaptor.AdaptorException; |
|||
import org.thingsboard.server.common.transport.auth.DeviceAuthService; |
|||
import org.thingsboard.server.common.transport.quota.QuotaService; |
|||
import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor; |
|||
import org.thingsboard.server.transport.coap.session.CoapExchangeObserverProxy; |
|||
import org.thingsboard.server.transport.coap.session.CoapSessionCtx; |
|||
import org.springframework.util.ReflectionUtils; |
|||
|
|||
@Slf4j |
|||
public class CoapTransportResource extends CoapResource { |
|||
// coap://localhost:port/api/v1/DEVICE_TOKEN/[attributes|telemetry|rpc[/requestId]]
|
|||
private static final int ACCESS_TOKEN_POSITION = 3; |
|||
private static final int FEATURE_TYPE_POSITION = 4; |
|||
private static final int REQUEST_ID_POSITION = 5; |
|||
|
|||
private final CoapTransportAdaptor adaptor; |
|||
private final SessionMsgProcessor processor; |
|||
private final DeviceAuthService authService; |
|||
private final QuotaService quotaService; |
|||
private final Field observerField; |
|||
private final long timeout; |
|||
|
|||
public CoapTransportResource(SessionMsgProcessor processor, DeviceAuthService authService, CoapTransportAdaptor adaptor, String name, |
|||
long timeout, QuotaService quotaService) { |
|||
super(name); |
|||
this.processor = processor; |
|||
this.authService = authService; |
|||
this.quotaService = quotaService; |
|||
this.adaptor = adaptor; |
|||
this.timeout = timeout; |
|||
// This is important to turn off existing observable logic in
|
|||
// CoapResource. We will have our own observe monitoring due to 1:1
|
|||
// observe relationship.
|
|||
this.setObservable(false); |
|||
observerField = ReflectionUtils.findField(Exchange.class, "observer"); |
|||
observerField.setAccessible(true); |
|||
} |
|||
|
|||
@Override |
|||
public void handleGET(CoapExchange exchange) { |
|||
if(quotaService.isQuotaExceeded(exchange.getSourceAddress().getHostAddress())) { |
|||
log.warn("COAP Quota exceeded for [{}:{}] . Disconnect", exchange.getSourceAddress().getHostAddress(), exchange.getSourcePort()); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
return; |
|||
} |
|||
|
|||
Optional<FeatureType> featureType = getFeatureType(exchange.advanced().getRequest()); |
|||
if (!featureType.isPresent()) { |
|||
log.trace("Missing feature type parameter"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} else if (featureType.get() == FeatureType.TELEMETRY) { |
|||
log.trace("Can't fetch/subscribe to timeseries updates"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} else if (exchange.getRequestOptions().hasObserve()) { |
|||
processExchangeGetRequest(exchange, featureType.get()); |
|||
} else if (featureType.get() == FeatureType.ATTRIBUTES) { |
|||
processRequest(exchange, SessionMsgType.GET_ATTRIBUTES_REQUEST); |
|||
} else { |
|||
log.trace("Invalid feature type parameter"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} |
|||
} |
|||
|
|||
private void processExchangeGetRequest(CoapExchange exchange, FeatureType featureType) { |
|||
boolean unsubscribe = exchange.getRequestOptions().getObserve() == 1; |
|||
SessionMsgType sessionMsgType; |
|||
if (featureType == FeatureType.RPC) { |
|||
sessionMsgType = unsubscribe ? SessionMsgType.UNSUBSCRIBE_RPC_COMMANDS_REQUEST : SessionMsgType.SUBSCRIBE_RPC_COMMANDS_REQUEST; |
|||
} else { |
|||
sessionMsgType = unsubscribe ? SessionMsgType.UNSUBSCRIBE_ATTRIBUTES_REQUEST : SessionMsgType.SUBSCRIBE_ATTRIBUTES_REQUEST; |
|||
} |
|||
Optional<SessionId> sessionId = processRequest(exchange, sessionMsgType); |
|||
if (sessionId.isPresent()) { |
|||
if (exchange.getRequestOptions().getObserve() == 1) { |
|||
exchange.respond(ResponseCode.VALID); |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void handlePOST(CoapExchange exchange) { |
|||
if(quotaService.isQuotaExceeded(exchange.getSourceAddress().getHostAddress())) { |
|||
log.warn("COAP Quota exceeded for [{}:{}] . Disconnect", exchange.getSourceAddress().getHostAddress(), exchange.getSourcePort()); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
return; |
|||
} |
|||
|
|||
Optional<FeatureType> featureType = getFeatureType(exchange.advanced().getRequest()); |
|||
if (!featureType.isPresent()) { |
|||
log.trace("Missing feature type parameter"); |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
} else { |
|||
switch (featureType.get()) { |
|||
case ATTRIBUTES: |
|||
processRequest(exchange, SessionMsgType.POST_ATTRIBUTES_REQUEST); |
|||
break; |
|||
case TELEMETRY: |
|||
processRequest(exchange, SessionMsgType.POST_TELEMETRY_REQUEST); |
|||
break; |
|||
case RPC: |
|||
Optional<Integer> requestId = getRequestId(exchange.advanced().getRequest()); |
|||
if (requestId.isPresent()) { |
|||
processRequest(exchange, SessionMsgType.TO_DEVICE_RPC_RESPONSE); |
|||
} else { |
|||
processRequest(exchange, SessionMsgType.TO_SERVER_RPC_REQUEST); |
|||
} |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
private Optional<SessionId> processRequest(CoapExchange exchange, SessionMsgType type) { |
|||
log.trace("Processing {}", exchange.advanced().getRequest()); |
|||
exchange.accept(); |
|||
Exchange advanced = exchange.advanced(); |
|||
Request request = advanced.getRequest(); |
|||
|
|||
Optional<DeviceCredentialsFilter> credentials = decodeCredentials(request); |
|||
if (!credentials.isPresent()) { |
|||
exchange.respond(ResponseCode.BAD_REQUEST); |
|||
return Optional.empty(); |
|||
} |
|||
|
|||
CoapSessionCtx ctx = new CoapSessionCtx(exchange, adaptor, processor, authService, timeout); |
|||
|
|||
// if (!ctx.login(credentials.get())) {
|
|||
// exchange.respond(ResponseCode.UNAUTHORIZED);
|
|||
// return Optional.empty();
|
|||
// }
|
|||
|
|||
AdaptorToSessionActorMsg msg; |
|||
try { |
|||
switch (type) { |
|||
case GET_ATTRIBUTES_REQUEST: |
|||
case POST_ATTRIBUTES_REQUEST: |
|||
case POST_TELEMETRY_REQUEST: |
|||
case TO_DEVICE_RPC_RESPONSE: |
|||
case TO_SERVER_RPC_REQUEST: |
|||
ctx.setSessionType(SessionType.SYNC); |
|||
msg = adaptor.convertToActorMsg(ctx, type, request); |
|||
break; |
|||
case SUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
case SUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
ExchangeObserver systemObserver = (ExchangeObserver) observerField.get(advanced); |
|||
advanced.setObserver(new CoapExchangeObserverProxy(systemObserver, ctx)); |
|||
ctx.setSessionType(SessionType.ASYNC); |
|||
msg = adaptor.convertToActorMsg(ctx, type, request); |
|||
break; |
|||
case UNSUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
case UNSUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
ctx.setSessionType(SessionType.ASYNC); |
|||
msg = adaptor.convertToActorMsg(ctx, type, request); |
|||
break; |
|||
default: |
|||
log.trace("[{}] Unsupported msg type: {}", ctx.getSessionId(), type); |
|||
throw new IllegalArgumentException("Unsupported msg type: " + type); |
|||
} |
|||
log.trace("Processing msg: {}", msg); |
|||
// processor.process(new BasicTransportToDeviceSessionActorMsg(ctx.getDevice(), msg));
|
|||
} catch (AdaptorException e) { |
|||
log.debug("Failed to decode payload {}", e); |
|||
exchange.respond(ResponseCode.BAD_REQUEST, e.getMessage()); |
|||
return Optional.empty(); |
|||
} catch (IllegalArgumentException | IllegalAccessException e) { |
|||
log.debug("Failed to process payload {}", e); |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR, e.getMessage()); |
|||
} |
|||
return Optional.of(ctx.getSessionId()); |
|||
} |
|||
|
|||
private Optional<DeviceCredentialsFilter> decodeCredentials(Request request) { |
|||
List<String> uriPath = request.getOptions().getUriPath(); |
|||
DeviceCredentialsFilter credentials = null; |
|||
if (uriPath.size() >= ACCESS_TOKEN_POSITION) { |
|||
credentials = new DeviceTokenCredentials(uriPath.get(ACCESS_TOKEN_POSITION - 1)); |
|||
} |
|||
return Optional.ofNullable(credentials); |
|||
} |
|||
|
|||
private Optional<FeatureType> getFeatureType(Request request) { |
|||
List<String> uriPath = request.getOptions().getUriPath(); |
|||
try { |
|||
if (uriPath.size() >= FEATURE_TYPE_POSITION) { |
|||
return Optional.of(FeatureType.valueOf(uriPath.get(FEATURE_TYPE_POSITION - 1).toUpperCase())); |
|||
} |
|||
} catch (RuntimeException e) { |
|||
log.warn("Failed to decode feature type: {}", uriPath); |
|||
} |
|||
return Optional.empty(); |
|||
} |
|||
|
|||
public static Optional<Integer> getRequestId(Request request) { |
|||
List<String> uriPath = request.getOptions().getUriPath(); |
|||
try { |
|||
if (uriPath.size() >= REQUEST_ID_POSITION) { |
|||
return Optional.of(Integer.valueOf(uriPath.get(REQUEST_ID_POSITION - 1))); |
|||
} |
|||
} catch (RuntimeException e) { |
|||
log.warn("Failed to decode feature type: {}", uriPath); |
|||
} |
|||
return Optional.empty(); |
|||
} |
|||
|
|||
@Override |
|||
public Resource getChild(String name) { |
|||
return this; |
|||
} |
|||
|
|||
} |
|||
@ -1,25 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap.adaptors; |
|||
|
|||
import org.eclipse.californium.core.coap.Request; |
|||
import org.eclipse.californium.core.coap.Response; |
|||
import org.thingsboard.server.common.transport.TransportAdaptor; |
|||
import org.thingsboard.server.transport.coap.session.CoapSessionCtx; |
|||
|
|||
public interface CoapTransportAdaptor extends TransportAdaptor<CoapSessionCtx, Request, Response> { |
|||
|
|||
} |
|||
@ -1,260 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap.adaptors; |
|||
|
|||
import java.util.*; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.californium.core.coap.CoAP.ResponseCode; |
|||
import org.eclipse.californium.core.coap.Request; |
|||
import org.eclipse.californium.core.coap.Response; |
|||
import org.springframework.util.StringUtils; |
|||
import org.thingsboard.server.common.msg.core.*; |
|||
import org.thingsboard.server.common.msg.kv.AttributesKVMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionContext; |
|||
import org.thingsboard.server.common.msg.session.ex.ProcessingTimeoutException; |
|||
import org.thingsboard.server.common.transport.adaptor.AdaptorException; |
|||
import org.thingsboard.server.common.transport.adaptor.JsonConverter; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import com.google.gson.JsonObject; |
|||
import com.google.gson.JsonParser; |
|||
import com.google.gson.JsonSyntaxException; |
|||
import org.thingsboard.server.transport.coap.CoapTransportResource; |
|||
import org.thingsboard.server.transport.coap.session.CoapSessionCtx; |
|||
|
|||
@Component("JsonCoapAdaptor") |
|||
@Slf4j |
|||
public class JsonCoapAdaptor implements CoapTransportAdaptor { |
|||
|
|||
@Override |
|||
public AdaptorToSessionActorMsg convertToActorMsg(CoapSessionCtx ctx, SessionMsgType type, Request inbound) throws AdaptorException { |
|||
FromDeviceMsg msg = null; |
|||
switch (type) { |
|||
case POST_TELEMETRY_REQUEST: |
|||
msg = convertToTelemetryUploadRequest(ctx, inbound); |
|||
break; |
|||
case POST_ATTRIBUTES_REQUEST: |
|||
msg = convertToUpdateAttributesRequest(ctx, inbound); |
|||
break; |
|||
case GET_ATTRIBUTES_REQUEST: |
|||
msg = convertToGetAttributesRequest(ctx, inbound); |
|||
break; |
|||
case SUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
msg = new RpcSubscribeMsg(); |
|||
break; |
|||
case UNSUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
msg = new RpcUnsubscribeMsg(); |
|||
break; |
|||
case TO_DEVICE_RPC_RESPONSE: |
|||
msg = convertToDeviceRpcResponse(ctx, inbound); |
|||
break; |
|||
case TO_SERVER_RPC_REQUEST: |
|||
msg = convertToServerRpcRequest(ctx, inbound); |
|||
break; |
|||
case SUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
msg = new AttributesSubscribeMsg(); |
|||
break; |
|||
case UNSUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
msg = new AttributesUnsubscribeMsg(); |
|||
break; |
|||
default: |
|||
log.warn("[{}] Unsupported msg type: {}!", ctx.getSessionId(), type); |
|||
throw new AdaptorException(new IllegalArgumentException("Unsupported msg type: " + type + "!")); |
|||
} |
|||
return new BasicAdaptorToSessionActorMsg(ctx, msg); |
|||
} |
|||
|
|||
private FromDeviceMsg convertToDeviceRpcResponse(CoapSessionCtx ctx, Request inbound) throws AdaptorException { |
|||
Optional<Integer> requestId = CoapTransportResource.getRequestId(inbound); |
|||
String payload = validatePayload(ctx, inbound); |
|||
JsonObject response = new JsonParser().parse(payload).getAsJsonObject(); |
|||
return new ToDeviceRpcResponseMsg( |
|||
requestId.orElseThrow(() -> new AdaptorException("Request id is missing!")), |
|||
response.get("response").toString()); |
|||
} |
|||
|
|||
private FromDeviceMsg convertToServerRpcRequest(CoapSessionCtx ctx, Request inbound) throws AdaptorException { |
|||
|
|||
String payload = validatePayload(ctx, inbound); |
|||
|
|||
// return JsonConverter.convertToServerRpcRequest(new JsonParser().parse(payload), 0);
|
|||
return null; |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Response> convertToAdaptorMsg(CoapSessionCtx ctx, SessionActorToAdaptorMsg source) throws AdaptorException { |
|||
ToDeviceMsg msg = source.getMsg(); |
|||
switch (msg.getSessionMsgType()) { |
|||
case STATUS_CODE_RESPONSE: |
|||
case TO_DEVICE_RPC_RESPONSE_ACK: |
|||
return Optional.of(convertStatusCodeResponse((StatusCodeResponse) msg)); |
|||
case GET_ATTRIBUTES_RESPONSE: |
|||
return Optional.of(convertGetAttributesResponse((GetAttributesResponse) msg)); |
|||
case ATTRIBUTES_UPDATE_NOTIFICATION: |
|||
return Optional.of(convertNotificationResponse(ctx, (AttributesUpdateNotification) msg)); |
|||
case TO_DEVICE_RPC_REQUEST: |
|||
return Optional.of(convertToDeviceRpcRequest(ctx, (ToDeviceRpcRequestMsg) msg)); |
|||
case TO_SERVER_RPC_RESPONSE: |
|||
return Optional.of(convertToServerRpcResponse(ctx, (ToServerRpcResponseMsg) msg)); |
|||
case RULE_ENGINE_ERROR: |
|||
return Optional.of(convertToRuleEngineErrorResponse(ctx, (RuleEngineErrorMsg) msg)); |
|||
default: |
|||
log.warn("[{}] Unsupported msg type: {}!", source.getSessionId(), msg.getSessionMsgType()); |
|||
throw new AdaptorException(new IllegalArgumentException("Unsupported msg type: " + msg.getSessionMsgType() + "!")); |
|||
} |
|||
} |
|||
|
|||
private Response convertToRuleEngineErrorResponse(CoapSessionCtx ctx, RuleEngineErrorMsg msg) { |
|||
ResponseCode status = ResponseCode.INTERNAL_SERVER_ERROR; |
|||
switch (msg.getError()) { |
|||
case QUEUE_PUT_TIMEOUT: |
|||
status = ResponseCode.GATEWAY_TIMEOUT; |
|||
break; |
|||
default: |
|||
if (msg.getInSessionMsgType() == SessionMsgType.TO_SERVER_RPC_REQUEST) { |
|||
status = ResponseCode.BAD_REQUEST; |
|||
} |
|||
break; |
|||
} |
|||
Response response = new Response(status); |
|||
response.setPayload(JsonConverter.toErrorJson(msg.getErrorMsg()).toString()); |
|||
return response; |
|||
} |
|||
|
|||
private Response convertNotificationResponse(CoapSessionCtx ctx, AttributesUpdateNotification msg) { |
|||
return getObserveNotification(ctx, JsonConverter.toJson(msg.getData(), false)); |
|||
} |
|||
|
|||
private Response convertToDeviceRpcRequest(CoapSessionCtx ctx, ToDeviceRpcRequestMsg msg) { |
|||
return getObserveNotification(ctx, JsonConverter.toJson(msg, true)); |
|||
} |
|||
|
|||
private Response getObserveNotification(CoapSessionCtx ctx, JsonObject json) { |
|||
Response response = new Response(ResponseCode.CONTENT); |
|||
response.getOptions().setObserve(ctx.nextSeqNumber()); |
|||
response.setPayload(json.toString()); |
|||
return response; |
|||
} |
|||
|
|||
private AttributesUpdateRequest convertToUpdateAttributesRequest(SessionContext ctx, Request inbound) throws AdaptorException { |
|||
String payload = validatePayload(ctx, inbound); |
|||
try { |
|||
return JsonConverter.convertToAttributes(new JsonParser().parse(payload)); |
|||
} catch (IllegalStateException | JsonSyntaxException ex) { |
|||
throw new AdaptorException(ex); |
|||
} |
|||
} |
|||
|
|||
private FromDeviceMsg convertToGetAttributesRequest(SessionContext ctx, Request inbound) throws AdaptorException { |
|||
List<String> queryElements = inbound.getOptions().getUriQuery(); |
|||
if (queryElements != null && queryElements.size() > 0) { |
|||
Set<String> clientKeys = toKeys(ctx, queryElements, "clientKeys"); |
|||
Set<String> sharedKeys = toKeys(ctx, queryElements, "sharedKeys"); |
|||
return new BasicGetAttributesRequest(0, clientKeys, sharedKeys); |
|||
} else { |
|||
return new BasicGetAttributesRequest(0); |
|||
} |
|||
} |
|||
|
|||
private Set<String> toKeys(SessionContext ctx, List<String> queryElements, String attributeName) throws AdaptorException { |
|||
String keys = null; |
|||
for (String queryElement : queryElements) { |
|||
String[] queryItem = queryElement.split("="); |
|||
if (queryItem.length == 2 && queryItem[0].equals(attributeName)) { |
|||
keys = queryItem[1]; |
|||
} |
|||
} |
|||
if (keys != null && !StringUtils.isEmpty(keys)) { |
|||
return new HashSet<>(Arrays.asList(keys.split(","))); |
|||
} else { |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
private TelemetryUploadRequest convertToTelemetryUploadRequest(SessionContext ctx, Request inbound) throws AdaptorException { |
|||
String payload = validatePayload(ctx, inbound); |
|||
try { |
|||
return JsonConverter.convertToTelemetry(new JsonParser().parse(payload)); |
|||
} catch (IllegalStateException | JsonSyntaxException ex) { |
|||
throw new AdaptorException(ex); |
|||
} |
|||
} |
|||
|
|||
private Response convertStatusCodeResponse(StatusCodeResponse msg) { |
|||
if (msg.isSuccess()) { |
|||
Optional<Integer> code = msg.getData(); |
|||
if (code.isPresent() && code.get() == 200) { |
|||
return new Response(ResponseCode.VALID); |
|||
} else { |
|||
return new Response(ResponseCode.CREATED); |
|||
} |
|||
} else { |
|||
return convertError(msg.getError()); |
|||
} |
|||
} |
|||
|
|||
private String validatePayload(SessionContext ctx, Request inbound) throws AdaptorException { |
|||
String payload = inbound.getPayloadString(); |
|||
if (payload == null) { |
|||
log.warn("[{}] Payload is empty!", ctx.getSessionId()); |
|||
throw new AdaptorException(new IllegalArgumentException("Payload is empty!")); |
|||
} |
|||
return payload; |
|||
} |
|||
|
|||
private Response convertToServerRpcResponse(SessionContext ctx, ToServerRpcResponseMsg msg) { |
|||
if (msg.isSuccess()) { |
|||
Response response = new Response(ResponseCode.CONTENT); |
|||
// JsonElement result = JsonConverter.toJson(msg);
|
|||
// response.setPayload(result.toString());
|
|||
return response; |
|||
} else { |
|||
return convertError(Optional.of(new RuntimeException("Server RPC response is empty!"))); |
|||
} |
|||
} |
|||
|
|||
private Response convertGetAttributesResponse(GetAttributesResponse msg) { |
|||
if (msg.isSuccess()) { |
|||
Optional<AttributesKVMsg> payload = msg.getData(); |
|||
if (!payload.isPresent() || (payload.get().getClientAttributes().isEmpty() && payload.get().getSharedAttributes().isEmpty())) { |
|||
return new Response(ResponseCode.NOT_FOUND); |
|||
} else { |
|||
Response response = new Response(ResponseCode.CONTENT); |
|||
JsonObject result = JsonConverter.toJson(payload.get(), false); |
|||
response.setPayload(result.toString()); |
|||
return response; |
|||
} |
|||
} else { |
|||
return convertError(msg.getError()); |
|||
} |
|||
} |
|||
|
|||
private Response convertError(Optional<Exception> exception) { |
|||
if (exception.isPresent()) { |
|||
log.warn("Converting exception: {}", exception.get().getMessage(), exception.get()); |
|||
if (exception.get() instanceof ProcessingTimeoutException) { |
|||
return new Response(ResponseCode.SERVICE_UNAVAILABLE); |
|||
} else { |
|||
return new Response(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} else { |
|||
return new Response(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,38 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap.session; |
|||
|
|||
import org.eclipse.californium.core.network.Exchange; |
|||
import org.eclipse.californium.core.network.ExchangeObserver; |
|||
|
|||
public class CoapExchangeObserverProxy implements ExchangeObserver { |
|||
|
|||
private final ExchangeObserver proxy; |
|||
private final CoapSessionCtx ctx; |
|||
|
|||
public CoapExchangeObserverProxy(ExchangeObserver proxy, CoapSessionCtx ctx) { |
|||
super(); |
|||
this.proxy = proxy; |
|||
this.ctx = ctx; |
|||
} |
|||
|
|||
@Override |
|||
public void completed(Exchange exchange) { |
|||
proxy.completed(exchange); |
|||
ctx.close(); |
|||
} |
|||
|
|||
} |
|||
@ -1,127 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap.session; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.californium.core.coap.CoAP.ResponseCode; |
|||
import org.eclipse.californium.core.coap.Request; |
|||
import org.eclipse.californium.core.coap.Response; |
|||
import org.eclipse.californium.core.server.resources.CoapExchange; |
|||
import org.thingsboard.server.common.msg.session.ex.SessionException; |
|||
import org.thingsboard.server.common.transport.SessionMsgProcessor; |
|||
import org.thingsboard.server.common.transport.adaptor.AdaptorException; |
|||
import org.thingsboard.server.common.transport.auth.DeviceAuthService; |
|||
import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext; |
|||
import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor; |
|||
|
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
@Slf4j |
|||
public class CoapSessionCtx extends DeviceAwareSessionContext { |
|||
|
|||
private final SessionId sessionId; |
|||
private final CoapExchange exchange; |
|||
private final CoapTransportAdaptor adaptor; |
|||
private final String token; |
|||
private final long timeout; |
|||
private SessionType sessionType; |
|||
private final AtomicInteger seqNumber = new AtomicInteger(2); |
|||
|
|||
public CoapSessionCtx(CoapExchange exchange, CoapTransportAdaptor adaptor, SessionMsgProcessor processor, DeviceAuthService authService, long timeout) { |
|||
super(); |
|||
Request request = exchange.advanced().getRequest(); |
|||
this.token = request.getTokenString(); |
|||
this.sessionId = new CoapSessionId(request.getSource().getHostAddress(), request.getSourcePort(), this.token); |
|||
this.exchange = exchange; |
|||
this.adaptor = adaptor; |
|||
this.timeout = timeout; |
|||
} |
|||
|
|||
|
|||
@Override |
|||
public void onMsg(SessionActorToAdaptorMsg msg) throws SessionException { |
|||
try { |
|||
adaptor.convertToAdaptorMsg(this, msg).ifPresent(this::pushToNetwork); |
|||
} catch (AdaptorException e) { |
|||
logAndWrap(e); |
|||
} |
|||
} |
|||
|
|||
private void pushToNetwork(Response response) { |
|||
exchange.respond(response); |
|||
} |
|||
|
|||
private void logAndWrap(AdaptorException e) throws SessionException { |
|||
log.warn("Failed to convert msg: {}", e.getMessage(), e); |
|||
throw new SessionException(e); |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(SessionCtrlMsg msg) throws SessionException { |
|||
log.debug("[{}] onCtrl: {}", sessionId, msg); |
|||
if (msg instanceof SessionCloseMsg) { |
|||
onSessionClose((SessionCloseMsg) msg); |
|||
} |
|||
} |
|||
|
|||
private void onSessionClose(SessionCloseMsg msg) { |
|||
if (msg.isTimeout()) { |
|||
exchange.respond(ResponseCode.SERVICE_UNAVAILABLE); |
|||
} else if (msg.isCredentialsRevoked()) { |
|||
exchange.respond(ResponseCode.UNAUTHORIZED); |
|||
} else { |
|||
exchange.respond(ResponseCode.INTERNAL_SERVER_ERROR); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public SessionId getSessionId() { |
|||
return sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "CoapSessionCtx [sessionId=" + sessionId + "]"; |
|||
} |
|||
|
|||
@Override |
|||
public boolean isClosed() { |
|||
return exchange.advanced().isComplete() || exchange.advanced().isTimedOut(); |
|||
} |
|||
|
|||
public void close() { |
|||
log.info("[{}] Closing processing context. Timeout: {}", sessionId, exchange.advanced().isTimedOut()); |
|||
// processor.process(exchange.advanced().isTimedOut() ? SessionCloseMsg.onTimeout(sessionId) : SessionCloseMsg.onError(sessionId));
|
|||
} |
|||
|
|||
@Override |
|||
public long getTimeout() { |
|||
return timeout; |
|||
} |
|||
|
|||
public void setSessionType(SessionType sessionType) { |
|||
this.sessionType = sessionType; |
|||
} |
|||
|
|||
@Override |
|||
public SessionType getSessionType() { |
|||
return sessionType; |
|||
} |
|||
|
|||
public int nextSeqNumber() { |
|||
return seqNumber.getAndIncrement(); |
|||
} |
|||
} |
|||
@ -1,75 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap.session; |
|||
|
|||
public final class CoapSessionId implements SessionId { |
|||
|
|||
private final String clientAddress; |
|||
private final int clientPort; |
|||
private final String token; |
|||
|
|||
public CoapSessionId(String host, int port, String token) { |
|||
super(); |
|||
this.clientAddress = host; |
|||
this.clientPort = port; |
|||
this.token = token; |
|||
} |
|||
|
|||
@Override |
|||
public int hashCode() { |
|||
final int prime = 31; |
|||
int result = 1; |
|||
result = prime * result + ((clientAddress == null) ? 0 : clientAddress.hashCode()); |
|||
result = prime * result + clientPort; |
|||
result = prime * result + ((token == null) ? 0 : token.hashCode()); |
|||
return result; |
|||
} |
|||
|
|||
@Override |
|||
public boolean equals(Object obj) { |
|||
if (this == obj) |
|||
return true; |
|||
if (obj == null) |
|||
return false; |
|||
if (getClass() != obj.getClass()) |
|||
return false; |
|||
CoapSessionId other = (CoapSessionId) obj; |
|||
if (clientAddress == null) { |
|||
if (other.clientAddress != null) |
|||
return false; |
|||
} else if (!clientAddress.equals(other.clientAddress)) |
|||
return false; |
|||
if (clientPort != other.clientPort) |
|||
return false; |
|||
if (token == null) { |
|||
if (other.token != null) |
|||
return false; |
|||
} else if (!token.equals(other.token)) |
|||
return false; |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "CoapSessionId [clientAddress=" + clientAddress + ", clientPort=" + clientPort + ", token=" + token + "]"; |
|||
} |
|||
|
|||
@Override |
|||
public String toUidStr() { |
|||
return clientAddress + ":" + clientPort + ":" + token; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!-- |
|||
|
|||
Copyright © 2016-2018 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. |
|||
|
|||
--> |
|||
<!DOCTYPE configuration> |
|||
<configuration scan="true" scanPeriod="10 seconds"> |
|||
|
|||
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> |
|||
<encoder> |
|||
<pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern> |
|||
</encoder> |
|||
</appender> |
|||
|
|||
<logger name="org.thingsboard.server" level="TRACE" /> |
|||
|
|||
<root level="INFO"> |
|||
<appender-ref ref="STDOUT"/> |
|||
</root> |
|||
|
|||
</configuration> |
|||
@ -0,0 +1,68 @@ |
|||
# |
|||
# Copyright © 2016-2018 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. |
|||
# |
|||
|
|||
spring.main.web-environment: false |
|||
spring.main.web-application-type: none |
|||
|
|||
# MQTT server parameters |
|||
transport: |
|||
coap: |
|||
bind_address: "${COAP_BIND_ADDRESS:0.0.0.0}" |
|||
bind_port: "${COAP_BIND_PORT:5683}" |
|||
timeout: "${COAP_TIMEOUT:10000}" |
|||
|
|||
#Quota parameters |
|||
quota: |
|||
host: |
|||
# Max allowed number of API requests in interval for single host |
|||
limit: "${QUOTA_HOST_LIMIT:10000}" |
|||
# Interval duration |
|||
intervalMs: "${QUOTA_HOST_INTERVAL_MS:60000}" |
|||
# Maximum silence duration for host after which Host removed from QuotaService. Must be bigger than intervalMs |
|||
ttlMs: "${QUOTA_HOST_TTL_MS:60000}" |
|||
# Interval for scheduled task that cleans expired records. TTL is used for expiring |
|||
cleanPeriodMs: "${QUOTA_HOST_CLEAN_PERIOD_MS:300000}" |
|||
# Enable Host API Limits |
|||
enabled: "${QUOTA_HOST_ENABLED:true}" |
|||
# Array of whitelist hosts |
|||
whitelist: "${QUOTA_HOST_WHITELIST:localhost,127.0.0.1}" |
|||
# Array of blacklist hosts |
|||
blacklist: "${QUOTA_HOST_BLACKLIST:}" |
|||
log: |
|||
topSize: 10 |
|||
intervalMin: 2 |
|||
|
|||
kafka: |
|||
enabled: true |
|||
bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}" |
|||
acks: "${TB_KAFKA_ACKS:all}" |
|||
retries: "${TB_KAFKA_RETRIES:1}" |
|||
batch.size: "${TB_KAFKA_BATCH_SIZE:16384}" |
|||
linger.ms: "${TB_KAFKA_LINGER_MS:1}" |
|||
buffer.memory: "${TB_BUFFER_MEMORY:33554432}" |
|||
transport_api: |
|||
requests_topic: "${TB_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}" |
|||
responses_topic: "${TB_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}" |
|||
max_pending_requests: "${TB_TRANSPORT_MAX_PENDING_REQUESTS:10000}" |
|||
max_requests_timeout: "${TB_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}" |
|||
response_poll_interval: "${TB_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}" |
|||
response_auto_commit_interval: "${TB_TRANSPORT_RESPONSE_AUTO_COMMIT_INTERVAL_MS:100}" |
|||
rule_engine: |
|||
topic: "${TB_RULE_ENGINE_TOPIC:tb.rule-engine}" |
|||
notifications: |
|||
topic: "${TB_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" |
|||
poll_interval: "${TB_TRANSPORT_NOTIFICATIONS_POLL_INTERVAL_MS:25}" |
|||
auto_commit_interval: "${TB_TRANSPORT_NOTIFICATIONS_AUTO_COMMIT_INTERVAL_MS:100}" |
|||
@ -0,0 +1,6 @@ |
|||
#!/bin/sh |
|||
|
|||
chown -R ${pkg.name}: ${pkg.logFolder} |
|||
chown -R ${pkg.name}: ${pkg.installFolder} |
|||
update-rc.d ${pkg.name} defaults |
|||
|
|||
@ -0,0 +1,3 @@ |
|||
#!/bin/sh |
|||
|
|||
update-rc.d -f ${pkg.name} remove |
|||
@ -0,0 +1,18 @@ |
|||
#!/bin/sh |
|||
|
|||
if ! getent group ${pkg.name} >/dev/null; then |
|||
addgroup --system ${pkg.name} |
|||
fi |
|||
|
|||
if ! getent passwd ${pkg.name} >/dev/null; then |
|||
adduser --quiet \ |
|||
--system \ |
|||
--ingroup ${pkg.name} \ |
|||
--quiet \ |
|||
--disabled-login \ |
|||
--disabled-password \ |
|||
--home ${pkg.installFolder} \ |
|||
--no-create-home \ |
|||
-gecos "Thingsboard application" \ |
|||
${pkg.name} |
|||
fi |
|||
@ -0,0 +1,5 @@ |
|||
#!/bin/sh |
|||
|
|||
if [ -e /var/run/${pkg.name}/${pkg.name}.pid ]; then |
|||
service ${pkg.name} stop |
|||
fi |
|||
@ -0,0 +1,9 @@ |
|||
#!/bin/sh |
|||
|
|||
chown -R ${pkg.name}: ${pkg.logFolder} |
|||
chown -R ${pkg.name}: ${pkg.installFolder} |
|||
|
|||
if [ $1 -eq 1 ] ; then |
|||
# Initial installation |
|||
systemctl --no-reload enable ${pkg.name}.service >/dev/null 2>&1 || : |
|||
fi |
|||
@ -0,0 +1,6 @@ |
|||
#!/bin/sh |
|||
|
|||
if [ $1 -ge 1 ] ; then |
|||
# Package upgrade, not uninstall |
|||
systemctl try-restart ${pkg.name}.service >/dev/null 2>&1 || : |
|||
fi |
|||
@ -0,0 +1,6 @@ |
|||
#!/bin/sh |
|||
|
|||
getent group ${pkg.name} >/dev/null || groupadd -r ${pkg.name} |
|||
getent passwd ${pkg.name} >/dev/null || \ |
|||
useradd -d ${pkg.installFolder} -g ${pkg.name} -M -r ${pkg.name} -s /sbin/nologin \ |
|||
-c "Thingsboard application" |
|||
@ -0,0 +1,6 @@ |
|||
#!/bin/sh |
|||
|
|||
if [ $1 -eq 0 ] ; then |
|||
# Package removal, not upgrade |
|||
systemctl --no-reload disable --now ${pkg.name}.service > /dev/null 2>&1 || : |
|||
fi |
|||
@ -0,0 +1,11 @@ |
|||
[Unit] |
|||
Description=${pkg.name} |
|||
After=syslog.target |
|||
|
|||
[Service] |
|||
User=${pkg.name} |
|||
ExecStart=${pkg.installFolder}/bin/${pkg.name}.jar |
|||
SuccessExitStatus=143 |
|||
|
|||
[Install] |
|||
WantedBy=multi-user.target |
|||
@ -0,0 +1,87 @@ |
|||
@ECHO OFF |
|||
|
|||
setlocal ENABLEEXTENSIONS |
|||
|
|||
@ECHO Detecting Java version installed. |
|||
:CHECK_JAVA_64 |
|||
@ECHO Detecting if it is 64 bit machine |
|||
set KEY_NAME="HKEY_LOCAL_MACHINE\Software\Wow6432Node\JavaSoft\Java Runtime Environment" |
|||
set VALUE_NAME=CurrentVersion |
|||
|
|||
FOR /F "usebackq skip=2 tokens=1-3" %%A IN (`REG QUERY %KEY_NAME% /v %VALUE_NAME% 2^>nul`) DO ( |
|||
set ValueName=%%A |
|||
set ValueType=%%B |
|||
set ValueValue=%%C |
|||
) |
|||
@ECHO CurrentVersion %ValueValue% |
|||
|
|||
SET KEY_NAME="%KEY_NAME:~1,-1%\%ValueValue%" |
|||
SET VALUE_NAME=JavaHome |
|||
|
|||
if defined ValueName ( |
|||
FOR /F "usebackq skip=2 tokens=1,2*" %%A IN (`REG QUERY %KEY_NAME% /v %VALUE_NAME% 2^>nul`) DO ( |
|||
set ValueName2=%%A |
|||
set ValueType2=%%B |
|||
set JRE_PATH2=%%C |
|||
|
|||
if defined ValueName2 ( |
|||
set ValueName = %ValueName2% |
|||
set ValueType = %ValueType2% |
|||
set ValueValue = %JRE_PATH2% |
|||
) |
|||
) |
|||
) |
|||
|
|||
IF NOT "%JRE_PATH2%" == "" GOTO JAVA_INSTALLED |
|||
|
|||
:CHECK_JAVA_32 |
|||
@ECHO Detecting if it is 32 bit machine |
|||
set KEY_NAME="HKEY_LOCAL_MACHINE\Software\JavaSoft\Java Runtime Environment" |
|||
set VALUE_NAME=CurrentVersion |
|||
|
|||
FOR /F "usebackq skip=2 tokens=1-3" %%A IN (`REG QUERY %KEY_NAME% /v %VALUE_NAME% 2^>nul`) DO ( |
|||
set ValueName=%%A |
|||
set ValueType=%%B |
|||
set ValueValue=%%C |
|||
) |
|||
@ECHO CurrentVersion %ValueValue% |
|||
|
|||
SET KEY_NAME="%KEY_NAME:~1,-1%\%ValueValue%" |
|||
SET VALUE_NAME=JavaHome |
|||
|
|||
if defined ValueName ( |
|||
FOR /F "usebackq skip=2 tokens=1,2*" %%A IN (`REG QUERY %KEY_NAME% /v %VALUE_NAME% 2^>nul`) DO ( |
|||
set ValueName2=%%A |
|||
set ValueType2=%%B |
|||
set JRE_PATH2=%%C |
|||
|
|||
if defined ValueName2 ( |
|||
set ValueName = %ValueName2% |
|||
set ValueType = %ValueType2% |
|||
set ValueValue = %JRE_PATH2% |
|||
) |
|||
) |
|||
) |
|||
|
|||
IF "%JRE_PATH2%" == "" GOTO JAVA_NOT_INSTALLED |
|||
|
|||
:JAVA_INSTALLED |
|||
|
|||
@ECHO Java 1.8 found! |
|||
@ECHO Installing ${pkg.name} ... |
|||
|
|||
%BASE%${pkg.name}.exe install |
|||
|
|||
@ECHO ${pkg.name} installed successfully! |
|||
|
|||
GOTO END |
|||
|
|||
:JAVA_NOT_INSTALLED |
|||
@ECHO Java 1.8 or above is not installed |
|||
@ECHO Please go to https://java.com/ and install Java. Then retry installation. |
|||
PAUSE |
|||
GOTO END |
|||
|
|||
:END |
|||
|
|||
|
|||
@ -0,0 +1,36 @@ |
|||
<service> |
|||
<id>${pkg.name}</id> |
|||
<name>${project.name}</name> |
|||
<description>${project.description}</description> |
|||
<workingdirectory>%BASE%\conf</workingdirectory> |
|||
<logpath>${pkg.winWrapperLogFolder}</logpath> |
|||
<logmode>rotate</logmode> |
|||
<env name="LOADER_PATH" value="%BASE%\conf" /> |
|||
<executable>java</executable> |
|||
<startargument>-Xloggc:%BASE%\logs\gc.log</startargument> |
|||
<startargument>-XX:+HeapDumpOnOutOfMemoryError</startargument> |
|||
<startargument>-XX:+PrintGCDetails</startargument> |
|||
<startargument>-XX:+PrintGCDateStamps</startargument> |
|||
<startargument>-XX:+PrintHeapAtGC</startargument> |
|||
<startargument>-XX:+PrintTenuringDistribution</startargument> |
|||
<startargument>-XX:+PrintGCApplicationStoppedTime</startargument> |
|||
<startargument>-XX:+UseGCLogFileRotation</startargument> |
|||
<startargument>-XX:NumberOfGCLogFiles=10</startargument> |
|||
<startargument>-XX:GCLogFileSize=10M</startargument> |
|||
<startargument>-XX:-UseBiasedLocking</startargument> |
|||
<startargument>-XX:+UseTLAB</startargument> |
|||
<startargument>-XX:+ResizeTLAB</startargument> |
|||
<startargument>-XX:+PerfDisableSharedMem</startargument> |
|||
<startargument>-XX:+UseCondCardMark</startargument> |
|||
<startargument>-XX:CMSWaitDuration=10000</startargument> |
|||
<startargument>-XX:+UseParNewGC</startargument> |
|||
<startargument>-XX:+UseConcMarkSweepGC</startargument> |
|||
<startargument>-XX:+CMSParallelRemarkEnabled</startargument> |
|||
<startargument>-XX:+CMSParallelInitialMarkEnabled</startargument> |
|||
<startargument>-XX:+CMSEdenChunksRecordAlways</startargument> |
|||
<startargument>-XX:CMSInitiatingOccupancyFraction=75</startargument> |
|||
<startargument>-XX:+UseCMSInitiatingOccupancyOnly</startargument> |
|||
<startargument>-jar</startargument> |
|||
<startargument>%BASE%\lib\${pkg.name}.jar</startargument> |
|||
|
|||
</service> |
|||
@ -0,0 +1,9 @@ |
|||
@ECHO OFF |
|||
|
|||
@ECHO Stopping ${pkg.name} ... |
|||
net stop ${pkg.name} |
|||
|
|||
@ECHO Uninstalling ${pkg.name} ... |
|||
%~dp0${pkg.name}.exe uninstall |
|||
|
|||
@ECHO DONE. |
|||
@ -1,197 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.californium.core.CoapClient; |
|||
import org.eclipse.californium.core.CoapResponse; |
|||
import org.eclipse.californium.core.coap.CoAP.ResponseCode; |
|||
import org.eclipse.californium.core.coap.MediaTypeRegistry; |
|||
import org.junit.Assert; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.junit.runner.RunWith; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.context.annotation.Bean; |
|||
import org.springframework.context.annotation.Configuration; |
|||
import org.springframework.test.annotation.DirtiesContext; |
|||
import org.springframework.test.annotation.DirtiesContext.ClassMode; |
|||
import org.springframework.test.context.junit4.SpringRunner; |
|||
import org.thingsboard.server.common.data.Device; |
|||
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.common.data.security.DeviceCredentialsFilter; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentialsType; |
|||
import org.thingsboard.server.common.data.security.DeviceTokenCredentials; |
|||
import org.thingsboard.server.common.msg.session.*; |
|||
import org.thingsboard.server.common.transport.SessionMsgProcessor; |
|||
import org.thingsboard.server.common.transport.auth.DeviceAuthResult; |
|||
import org.thingsboard.server.common.transport.auth.DeviceAuthService; |
|||
import org.thingsboard.server.common.transport.quota.host.HostRequestsQuotaService; |
|||
|
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
|
|||
@RunWith(SpringRunner.class) |
|||
@DirtiesContext(classMode = ClassMode.BEFORE_CLASS) |
|||
@Slf4j |
|||
public class CoapServerTest { |
|||
|
|||
private static final int TEST_PORT = 5555; |
|||
private static final String TELEMETRY_POST_MESSAGE = "[{\"key1\":\"value1\"}]"; |
|||
private static final String TEST_ATTRIBUTES_RESPONSE = "{\"key1\":\"value1\",\"key2\":42}"; |
|||
private static final String DEVICE1_TOKEN = "Device1Token"; |
|||
|
|||
@Configuration |
|||
public static class EchoCoapServerITConfiguration extends CoapServerTestConfiguration { |
|||
|
|||
@Bean |
|||
public static DeviceAuthService authService() { |
|||
return new DeviceAuthService() { |
|||
|
|||
private final DeviceId devId = new DeviceId(UUID.randomUUID()); |
|||
|
|||
@Override |
|||
public DeviceAuthResult process(DeviceCredentialsFilter credentials) { |
|||
if (credentials != null && credentials.getCredentialsType() == DeviceCredentialsType.ACCESS_TOKEN) { |
|||
DeviceTokenCredentials tokenCredentials = (DeviceTokenCredentials) credentials; |
|||
if (tokenCredentials.getCredentialsId().equals(DEVICE1_TOKEN)) { |
|||
return DeviceAuthResult.of(devId); |
|||
} |
|||
} |
|||
return DeviceAuthResult.of("Credentials are invalid!"); |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Device> findDeviceById(DeviceId deviceId) { |
|||
if (deviceId.equals(devId)) { |
|||
Device dev = new Device(); |
|||
dev.setId(devId); |
|||
dev.setTenantId(new TenantId(UUID.randomUUID())); |
|||
dev.setCustomerId(new CustomerId(UUID.randomUUID())); |
|||
return Optional.of(dev); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
} |
|||
}; |
|||
} |
|||
|
|||
@Bean |
|||
public static SessionMsgProcessor sessionMsgProcessor() { |
|||
return new SessionMsgProcessor() { |
|||
|
|||
// @Override
|
|||
// public void process(SessionAwareMsg toActorMsg) {
|
|||
// if (toActorMsg instanceof TransportToDeviceSessionActorMsg) {
|
|||
// AdaptorToSessionActorMsg sessionMsg = ((TransportToDeviceSessionActorMsg) toActorMsg).getSessionMsg();
|
|||
// try {
|
|||
// FromDeviceMsg deviceMsg = sessionMsg.getMsg();
|
|||
// ToDeviceMsg toDeviceMsg = null;
|
|||
// if (deviceMsg.getMsgType() == SessionMsgType.POST_TELEMETRY_REQUEST) {
|
|||
// toDeviceMsg = BasicStatusCodeResponse.onSuccess(deviceMsg.getMsgType(), BasicRequest.DEFAULT_REQUEST_ID);
|
|||
// } else if (deviceMsg.getMsgType() == SessionMsgType.GET_ATTRIBUTES_REQUEST) {
|
|||
// List<AttributeKvEntry> data = new ArrayList<>();
|
|||
// data.add(new BaseAttributeKvEntry(new StringDataEntry("key1", "value1"), System.currentTimeMillis()));
|
|||
// data.add(new BaseAttributeKvEntry(new LongDataEntry("key2", 42L), System.currentTimeMillis()));
|
|||
// BasicAttributeKVMsg kv = BasicAttributeKVMsg.fromClient(data);
|
|||
// toDeviceMsg = BasicGetAttributesResponse.onSuccess(deviceMsg.getMsgType(), BasicRequest.DEFAULT_REQUEST_ID, kv);
|
|||
// }
|
|||
// if (toDeviceMsg != null) {
|
|||
// sessionMsg.getSessionContext().onMsg(new BasicSessionActorToAdaptorMsg(sessionMsg.getSessionContext(), toDeviceMsg));
|
|||
// }
|
|||
// } catch (Exception e) {
|
|||
// e.printStackTrace();
|
|||
// }
|
|||
// }
|
|||
// }
|
|||
|
|||
@Override |
|||
public void onDeviceAdded(Device device) { |
|||
|
|||
} |
|||
|
|||
|
|||
}; |
|||
} |
|||
|
|||
@Bean |
|||
public static HostRequestsQuotaService quotaService() { |
|||
return new HostRequestsQuotaService(null, null, null, null, false); |
|||
} |
|||
} |
|||
|
|||
@Autowired |
|||
private CoapTransportService service; |
|||
|
|||
@Before |
|||
public void beforeTest() { |
|||
log.info("Service info: {}", service.toString()); |
|||
} |
|||
|
|||
@Test |
|||
public void testBadJsonTelemetryPostRequest() { |
|||
CoapClient client = new CoapClient(getBaseTestUrl() + DEVICE1_TOKEN + "/" + FeatureType.TELEMETRY.name().toLowerCase()); |
|||
CoapResponse response = client.setTimeout(6000).post("test", MediaTypeRegistry.APPLICATION_JSON); |
|||
Assert.assertEquals(ResponseCode.BAD_REQUEST, response.getCode()); |
|||
log.info("Response: {}, {}", response.getCode(), response.getResponseText()); |
|||
} |
|||
|
|||
@Test |
|||
public void testNoCredentialsPostRequest() { |
|||
CoapClient client = new CoapClient(getBaseTestUrl()); |
|||
CoapResponse response = client.setTimeout(6000).post(TELEMETRY_POST_MESSAGE, MediaTypeRegistry.APPLICATION_JSON); |
|||
Assert.assertEquals(ResponseCode.BAD_REQUEST, response.getCode()); |
|||
log.info("Response: {}, {}", response.getCode(), response.getResponseText()); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidJsonTelemetryPostRequest() { |
|||
CoapClient client = new CoapClient(getBaseTestUrl() + DEVICE1_TOKEN + "/" + FeatureType.TELEMETRY.name().toLowerCase()); |
|||
CoapResponse response = client.setTimeout(6000).post(TELEMETRY_POST_MESSAGE, MediaTypeRegistry.APPLICATION_JSON); |
|||
Assert.assertEquals(ResponseCode.CREATED, response.getCode()); |
|||
log.info("Response: {}, {}", response.getCode(), response.getResponseText()); |
|||
} |
|||
|
|||
@Test |
|||
public void testNoCredentialsAttributesGetRequest() { |
|||
CoapClient client = new CoapClient("coap://localhost:5555/api/v1?keys=key1,key2"); |
|||
CoapResponse response = client.setTimeout(6000).get(); |
|||
Assert.assertEquals(ResponseCode.BAD_REQUEST, response.getCode()); |
|||
} |
|||
|
|||
@Test |
|||
public void testNoKeysAttributesGetRequest() { |
|||
CoapClient client = new CoapClient(getBaseTestUrl() + DEVICE1_TOKEN + "/" + FeatureType.ATTRIBUTES.name().toLowerCase() + "?data=key1,key2"); |
|||
CoapResponse response = client.setTimeout(6000).get(); |
|||
Assert.assertEquals(ResponseCode.CONTENT, response.getCode()); |
|||
} |
|||
|
|||
@Test |
|||
public void testValidAttributesGetRequest() { |
|||
CoapClient client = new CoapClient(getBaseTestUrl() + DEVICE1_TOKEN + "/" + FeatureType.ATTRIBUTES.name().toLowerCase() + "?clientKeys=key1,key2"); |
|||
CoapResponse response = client.setTimeout(6000).get(); |
|||
Assert.assertEquals(ResponseCode.CONTENT, response.getCode()); |
|||
Assert.assertEquals(TEST_ATTRIBUTES_RESPONSE, response.getResponseText()); |
|||
log.info("Response: {}, {}", response.getCode(), response.getResponseText()); |
|||
} |
|||
|
|||
private String getBaseTestUrl() { |
|||
return "coap://localhost:" + TEST_PORT + "/api/v1/"; |
|||
} |
|||
|
|||
} |
|||
@ -1,35 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.transport.coap; |
|||
|
|||
import org.springframework.context.annotation.Bean; |
|||
import org.springframework.context.annotation.ComponentScan; |
|||
import org.springframework.context.annotation.Configuration; |
|||
import org.springframework.context.annotation.PropertySource; |
|||
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer; |
|||
import org.springframework.test.context.TestPropertySource; |
|||
|
|||
@Configuration |
|||
@ComponentScan({ "org.thingsboard.server.transport.coap" }) |
|||
@PropertySource("classpath:coap-transport-test.properties") |
|||
public class CoapServerTestConfiguration { |
|||
|
|||
@Bean |
|||
public static PropertySourcesPlaceholderConfigurer propertyConfig() { |
|||
return new PropertySourcesPlaceholderConfigurer(); |
|||
} |
|||
|
|||
} |
|||
@ -1,4 +0,0 @@ |
|||
coap.bind_address=0.0.0.0 |
|||
coap.bind_port=5555 |
|||
coap.adaptor=JsonCoapAdaptor |
|||
coap.timeout=10000 |
|||
Loading…
Reference in new issue