ShvaykaD
8 years ago
693 changed files with 31498 additions and 13495 deletions
File diff suppressed because one or more lines are too long
@ -0,0 +1,81 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_name; |
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_search_text; |
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_customer; |
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_entity_id; |
|||
|
|||
DROP TABLE IF EXISTS thingsboard.entity_views; |
|||
|
|||
CREATE TABLE IF NOT EXISTS thingsboard.entity_views ( |
|||
id timeuuid, |
|||
entity_id timeuuid, |
|||
entity_type text, |
|||
tenant_id timeuuid, |
|||
customer_id timeuuid, |
|||
name text, |
|||
keys text, |
|||
start_ts bigint, |
|||
end_ts bigint, |
|||
search_text text, |
|||
additional_info text, |
|||
PRIMARY KEY (id, entity_id, tenant_id, customer_id) |
|||
); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_name AS |
|||
SELECT * |
|||
from thingsboard.entity_views |
|||
WHERE tenant_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND name IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, name, id, customer_id, entity_id) |
|||
WITH CLUSTERING ORDER BY (name ASC, id DESC, customer_id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_search_text AS |
|||
SELECT * |
|||
from thingsboard.entity_views |
|||
WHERE tenant_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, search_text, id, customer_id, entity_id) |
|||
WITH CLUSTERING ORDER BY (search_text ASC, id DESC, customer_id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_customer AS |
|||
SELECT * |
|||
from thingsboard.entity_views |
|||
WHERE tenant_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, customer_id, search_text, id, entity_id) |
|||
WITH CLUSTERING ORDER BY (customer_id DESC, search_text ASC, id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_entity_id AS |
|||
SELECT * |
|||
from thingsboard.entity_views |
|||
WHERE tenant_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, entity_id, customer_id, search_text, id) |
|||
WITH CLUSTERING ORDER BY (entity_id DESC, customer_id DESC, search_text ASC, id DESC); |
|||
@ -0,0 +1,31 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
DROP TABLE IF EXISTS entity_views; |
|||
|
|||
CREATE TABLE IF NOT EXISTS entity_views ( |
|||
id varchar(31) NOT NULL CONSTRAINT entity_view_pkey PRIMARY KEY, |
|||
entity_id varchar(31), |
|||
entity_type varchar(255), |
|||
tenant_id varchar(31), |
|||
customer_id varchar(31), |
|||
name varchar(255), |
|||
keys varchar(255), |
|||
start_ts bigint, |
|||
end_ts bigint, |
|||
search_text varchar(255), |
|||
additional_info varchar |
|||
); |
|||
@ -0,0 +1,110 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_name; |
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_search_text; |
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_customer; |
|||
DROP MATERIALIZED VIEW IF EXISTS thingsboard.entity_view_by_tenant_and_entity_id; |
|||
|
|||
DROP TABLE IF EXISTS thingsboard.entity_views; |
|||
|
|||
CREATE TABLE IF NOT EXISTS thingsboard.entity_view ( |
|||
id timeuuid, |
|||
entity_id timeuuid, |
|||
entity_type text, |
|||
tenant_id timeuuid, |
|||
customer_id timeuuid, |
|||
name text, |
|||
type text, |
|||
keys text, |
|||
start_ts bigint, |
|||
end_ts bigint, |
|||
search_text text, |
|||
additional_info text, |
|||
PRIMARY KEY (id, entity_id, tenant_id, customer_id, type) |
|||
); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_name AS |
|||
SELECT * |
|||
from thingsboard.entity_view |
|||
WHERE tenant_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND type IS NOT NULL |
|||
AND name IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, name, id, customer_id, entity_id, type) |
|||
WITH CLUSTERING ORDER BY (name ASC, id DESC, customer_id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_search_text AS |
|||
SELECT * |
|||
from thingsboard.entity_view |
|||
WHERE tenant_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND type IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, search_text, id, customer_id, entity_id, type) |
|||
WITH CLUSTERING ORDER BY (search_text ASC, id DESC, customer_id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_by_type_and_search_text AS |
|||
SELECT * |
|||
from thingsboard.entity_view |
|||
WHERE tenant_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND type IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, type, search_text, id, customer_id, entity_id) |
|||
WITH CLUSTERING ORDER BY (type ASC, search_text ASC, id DESC, customer_id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_customer AS |
|||
SELECT * |
|||
from thingsboard.entity_view |
|||
WHERE tenant_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND type IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, customer_id, search_text, id, entity_id, type) |
|||
WITH CLUSTERING ORDER BY (customer_id DESC, search_text ASC, id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_customer_and_type AS |
|||
SELECT * |
|||
from thingsboard.entity_view |
|||
WHERE tenant_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND type IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, type, customer_id, search_text, id, entity_id) |
|||
WITH CLUSTERING ORDER BY (type ASC, customer_id DESC, search_text ASC, id DESC); |
|||
|
|||
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.entity_view_by_tenant_and_entity_id AS |
|||
SELECT * |
|||
from thingsboard.entity_view |
|||
WHERE tenant_id IS NOT NULL |
|||
AND customer_id IS NOT NULL |
|||
AND entity_id IS NOT NULL |
|||
AND type IS NOT NULL |
|||
AND search_text IS NOT NULL |
|||
AND id IS NOT NULL |
|||
PRIMARY KEY (tenant_id, entity_id, customer_id, search_text, id, type) |
|||
WITH CLUSTERING ORDER BY (entity_id DESC, customer_id DESC, search_text ASC, id DESC); |
|||
@ -0,0 +1,32 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
DROP TABLE IF EXISTS entity_views; |
|||
|
|||
CREATE TABLE IF NOT EXISTS entity_view ( |
|||
id varchar(31) NOT NULL CONSTRAINT entity_view_pkey PRIMARY KEY, |
|||
entity_id varchar(31), |
|||
entity_type varchar(255), |
|||
tenant_id varchar(31), |
|||
customer_id varchar(31), |
|||
type varchar(255), |
|||
name varchar(255), |
|||
keys varchar(255), |
|||
start_ts bigint, |
|||
end_ts bigint, |
|||
search_text varchar(255), |
|||
additional_info varchar |
|||
); |
|||
@ -0,0 +1,17 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
ALTER TABLE component_descriptor ADD UNIQUE (clazz); |
|||
@ -0,0 +1,39 @@ |
|||
/** |
|||
* 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.actors.device; |
|||
|
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public class DeviceActorCreator extends ContextBasedCreator<DeviceActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final TenantId tenantId; |
|||
private final DeviceId deviceId; |
|||
|
|||
public DeviceActorCreator(ActorSystemContext context, TenantId tenantId, DeviceId deviceId) { |
|||
super(context); |
|||
this.tenantId = tenantId; |
|||
this.deviceId = deviceId; |
|||
} |
|||
|
|||
@Override |
|||
public DeviceActor create() { |
|||
return new DeviceActor(context, tenantId, deviceId); |
|||
} |
|||
} |
|||
@ -1,40 +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.actors.device; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
/** |
|||
* Created by ashvayka on 17.04.18. |
|||
*/ |
|||
@Data |
|||
@AllArgsConstructor |
|||
public final class PendingSessionMsgData { |
|||
|
|||
private final SessionId sessionId; |
|||
private final Optional<ServerAddress> serverAddress; |
|||
private final SessionMsgType sessionMsgType; |
|||
private final int requestId; |
|||
private final boolean replyOnQueueAck; |
|||
private int ackMsgCount; |
|||
|
|||
} |
|||
@ -1,156 +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.actors.session; |
|||
|
|||
import akka.actor.ActorContext; |
|||
import akka.event.LoggingAdapter; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.core.AttributesSubscribeMsg; |
|||
import org.thingsboard.server.common.msg.core.ResponseMsg; |
|||
import org.thingsboard.server.common.msg.core.RpcSubscribeMsg; |
|||
import org.thingsboard.server.common.msg.core.SessionCloseMsg; |
|||
import org.thingsboard.server.common.msg.core.SessionOpenMsg; |
|||
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.session.BasicSessionActorToAdaptorMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ex.SessionException; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
|
|||
class ASyncMsgProcessor extends AbstractSessionActorMsgProcessor { |
|||
|
|||
private boolean firstMsg = true; |
|||
private Map<Integer, DeviceToDeviceActorMsg> pendingMap = new HashMap<>(); |
|||
private Optional<ServerAddress> currentTargetServer; |
|||
private boolean subscribedToAttributeUpdates; |
|||
private boolean subscribedToRpcCommands; |
|||
|
|||
public ASyncMsgProcessor(ActorSystemContext ctx, LoggingAdapter logger, SessionId sessionId) { |
|||
super(ctx, logger, sessionId); |
|||
} |
|||
|
|||
@Override |
|||
protected void processToDeviceActorMsg(ActorContext ctx, TransportToDeviceSessionActorMsg msg) { |
|||
updateSessionCtx(msg, SessionType.ASYNC); |
|||
DeviceToDeviceActorMsg pendingMsg = toDeviceMsg(msg); |
|||
FromDeviceMsg fromDeviceMsg = pendingMsg.getPayload(); |
|||
if (firstMsg) { |
|||
if (fromDeviceMsg.getMsgType() != SessionMsgType.SESSION_OPEN) { |
|||
toDeviceMsg(new SessionOpenMsg()).ifPresent(m -> forwardToAppActor(ctx, m)); |
|||
} |
|||
firstMsg = false; |
|||
} |
|||
switch (fromDeviceMsg.getMsgType()) { |
|||
case POST_TELEMETRY_REQUEST: |
|||
case POST_ATTRIBUTES_REQUEST: |
|||
FromDeviceRequestMsg requestMsg = (FromDeviceRequestMsg) fromDeviceMsg; |
|||
if (requestMsg.getRequestId() >= 0) { |
|||
logger.debug("[{}] Pending request {} registered", requestMsg.getRequestId(), requestMsg.getMsgType()); |
|||
//TODO: handle duplicates.
|
|||
pendingMap.put(requestMsg.getRequestId(), pendingMsg); |
|||
} |
|||
break; |
|||
case SUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
subscribedToAttributeUpdates = true; |
|||
break; |
|||
case UNSUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
subscribedToAttributeUpdates = false; |
|||
break; |
|||
case SUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
subscribedToRpcCommands = true; |
|||
break; |
|||
case UNSUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
subscribedToRpcCommands = false; |
|||
break; |
|||
default: |
|||
break; |
|||
} |
|||
currentTargetServer = forwardToAppActor(ctx, pendingMsg); |
|||
} |
|||
|
|||
@Override |
|||
public void processToDeviceMsg(ActorContext context, ToDeviceMsg msg) { |
|||
try { |
|||
if (msg.getSessionMsgType() != SessionMsgType.SESSION_CLOSE) { |
|||
switch (msg.getSessionMsgType()) { |
|||
case STATUS_CODE_RESPONSE: |
|||
case GET_ATTRIBUTES_RESPONSE: |
|||
ResponseMsg responseMsg = (ResponseMsg) msg; |
|||
if (responseMsg.getRequestId() >= 0) { |
|||
logger.debug("[{}] Pending request processed: {}", responseMsg.getRequestId(), responseMsg); |
|||
pendingMap.remove(responseMsg.getRequestId()); |
|||
} |
|||
break; |
|||
default: |
|||
break; |
|||
} |
|||
sessionCtx.onMsg(new BasicSessionActorToAdaptorMsg(this.sessionCtx, msg)); |
|||
} else { |
|||
sessionCtx.onMsg(org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg.onCredentialsRevoked(sessionCtx.getSessionId())); |
|||
} |
|||
} catch (SessionException e) { |
|||
logger.warning("Failed to push session response msg", e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void processTimeoutMsg(ActorContext context, SessionTimeoutMsg msg) { |
|||
// TODO Auto-generated method stub
|
|||
} |
|||
|
|||
@Override |
|||
protected void cleanupSession(ActorContext ctx) { |
|||
toDeviceMsg(new SessionCloseMsg()).ifPresent(m -> forwardToAppActor(ctx, m)); |
|||
} |
|||
|
|||
@Override |
|||
public void processClusterEvent(ActorContext context, ClusterEventMsg msg) { |
|||
if (pendingMap.size() > 0 || subscribedToAttributeUpdates || subscribedToRpcCommands) { |
|||
Optional<ServerAddress> newTargetServer = systemContext.getRoutingService().resolveById(getDeviceId()); |
|||
if (!newTargetServer.equals(currentTargetServer)) { |
|||
firstMsg = true; |
|||
currentTargetServer = newTargetServer; |
|||
pendingMap.values().forEach(v -> { |
|||
forwardToAppActor(context, v, currentTargetServer); |
|||
if (currentTargetServer.isPresent()) { |
|||
logger.debug("[{}] Forwarded msg to new server: {}", sessionId, currentTargetServer.get()); |
|||
} else { |
|||
logger.debug("[{}] Forwarded msg to local server.", sessionId); |
|||
} |
|||
}); |
|||
if (subscribedToAttributeUpdates) { |
|||
toDeviceMsg(new AttributesSubscribeMsg()).ifPresent(m -> forwardToAppActor(context, m, currentTargetServer)); |
|||
logger.debug("[{}] Forwarded attributes subscription.", sessionId); |
|||
} |
|||
if (subscribedToRpcCommands) { |
|||
toDeviceMsg(new RpcSubscribeMsg()).ifPresent(m -> forwardToAppActor(context, m, currentTargetServer)); |
|||
logger.debug("[{}] Forwarded rpc commands subscription.", sessionId); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,122 +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.actors.session; |
|||
|
|||
import akka.actor.ActorContext; |
|||
import akka.actor.ActorRef; |
|||
import akka.event.LoggingAdapter; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.device.BasicDeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.session.AdaptorToSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionContext; |
|||
import org.thingsboard.server.common.msg.session.SessionCtrlMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
abstract class AbstractSessionActorMsgProcessor extends AbstractContextAwareMsgProcessor { |
|||
|
|||
protected final SessionId sessionId; |
|||
protected SessionContext sessionCtx; |
|||
protected DeviceToDeviceActorMsg deviceToDeviceActorMsgPrototype; |
|||
|
|||
protected AbstractSessionActorMsgProcessor(ActorSystemContext ctx, LoggingAdapter logger, SessionId sessionId) { |
|||
super(ctx, logger); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
protected abstract void processToDeviceActorMsg(ActorContext ctx, TransportToDeviceSessionActorMsg msg); |
|||
|
|||
protected abstract void processTimeoutMsg(ActorContext context, SessionTimeoutMsg msg); |
|||
|
|||
protected abstract void processToDeviceMsg(ActorContext context, ToDeviceMsg msg); |
|||
|
|||
public abstract void processClusterEvent(ActorContext context, ClusterEventMsg msg); |
|||
|
|||
protected void processSessionCtrlMsg(ActorContext ctx, SessionCtrlMsg msg) { |
|||
if (msg instanceof SessionCloseMsg) { |
|||
cleanupSession(ctx); |
|||
terminateSession(ctx, sessionId); |
|||
} |
|||
} |
|||
|
|||
protected void cleanupSession(ActorContext ctx) { |
|||
} |
|||
|
|||
protected void updateSessionCtx(TransportToDeviceSessionActorMsg msg, SessionType type) { |
|||
sessionCtx = msg.getSessionMsg().getSessionContext(); |
|||
deviceToDeviceActorMsgPrototype = new BasicDeviceToDeviceActorMsg(msg, type); |
|||
} |
|||
|
|||
protected DeviceToDeviceActorMsg toDeviceMsg(TransportToDeviceSessionActorMsg msg) { |
|||
AdaptorToSessionActorMsg adaptorMsg = msg.getSessionMsg(); |
|||
return new BasicDeviceToDeviceActorMsg(deviceToDeviceActorMsgPrototype, adaptorMsg.getMsg()); |
|||
} |
|||
|
|||
protected Optional<DeviceToDeviceActorMsg> toDeviceMsg(FromDeviceMsg msg) { |
|||
if (deviceToDeviceActorMsgPrototype != null) { |
|||
return Optional.of(new BasicDeviceToDeviceActorMsg(deviceToDeviceActorMsgPrototype, msg)); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
} |
|||
|
|||
protected Optional<ServerAddress> forwardToAppActor(ActorContext ctx, DeviceToDeviceActorMsg toForward) { |
|||
Optional<ServerAddress> address = systemContext.getRoutingService().resolveById(toForward.getDeviceId()); |
|||
forwardToAppActor(ctx, toForward, address); |
|||
return address; |
|||
} |
|||
|
|||
protected Optional<ServerAddress> forwardToAppActorIfAddressChanged(ActorContext ctx, DeviceToDeviceActorMsg toForward, Optional<ServerAddress> oldAddress) { |
|||
|
|||
Optional<ServerAddress> newAddress = systemContext.getRoutingService().resolveById(toForward.getDeviceId()); |
|||
if (!newAddress.equals(oldAddress)) { |
|||
getAppActor().tell(new SendToClusterMsg(toForward.getDeviceId(), toForward |
|||
.toOtherAddress(systemContext.getRoutingService().getCurrentServer())), ctx.self()); |
|||
} |
|||
return newAddress; |
|||
} |
|||
|
|||
protected void forwardToAppActor(ActorContext ctx, DeviceToDeviceActorMsg toForward, Optional<ServerAddress> address) { |
|||
if (address.isPresent()) { |
|||
systemContext.getRpcService().tell(systemContext.getEncodingService().convertToProtoDataMessage(address.get(), |
|||
toForward.toOtherAddress(systemContext.getRoutingService().getCurrentServer()))); |
|||
} else { |
|||
getAppActor().tell(toForward, ctx.self()); |
|||
} |
|||
} |
|||
|
|||
public static void terminateSession(ActorContext ctx, SessionId sessionId) { |
|||
ctx.parent().tell(new SessionTerminationMsg(sessionId), ActorRef.noSender()); |
|||
ctx.stop(ctx.self()); |
|||
} |
|||
|
|||
public DeviceId getDeviceId() { |
|||
return deviceToDeviceActorMsgPrototype.getDeviceId(); |
|||
} |
|||
} |
|||
@ -1,143 +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.actors.session; |
|||
|
|||
import akka.actor.OneForOneStrategy; |
|||
import akka.actor.SupervisorStrategy; |
|||
import akka.event.Logging; |
|||
import akka.event.LoggingAdapter; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ContextAwareActor; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.core.ActorSystemToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionCtrlMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg; |
|||
import scala.concurrent.duration.Duration; |
|||
|
|||
public class SessionActor extends ContextAwareActor { |
|||
|
|||
private final LoggingAdapter logger = Logging.getLogger(getContext().system(), this); |
|||
|
|||
private final SessionId sessionId; |
|||
private AbstractSessionActorMsgProcessor processor; |
|||
|
|||
private SessionActor(ActorSystemContext systemContext, SessionId sessionId) { |
|||
super(systemContext); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public SupervisorStrategy supervisorStrategy() { |
|||
return new OneForOneStrategy(-1, Duration.Inf(), |
|||
throwable -> { |
|||
logger.error(throwable, "Unknown session error"); |
|||
if (throwable instanceof Error) { |
|||
return OneForOneStrategy.escalate(); |
|||
} else { |
|||
return OneForOneStrategy.resume(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
protected boolean process(TbActorMsg msg) { |
|||
switch (msg.getMsgType()) { |
|||
case TRANSPORT_TO_DEVICE_SESSION_ACTOR_MSG: |
|||
processTransportToSessionMsg((TransportToDeviceSessionActorMsg) msg); |
|||
break; |
|||
case ACTOR_SYSTEM_TO_DEVICE_SESSION_ACTOR_MSG: |
|||
processActorsToSessionMsg((ActorSystemToDeviceSessionActorMsg) msg); |
|||
break; |
|||
case SESSION_TIMEOUT_MSG: |
|||
processTimeoutMsg((SessionTimeoutMsg) msg); |
|||
break; |
|||
case SESSION_CTRL_MSG: |
|||
processSessionCloseMsg((SessionCtrlMsg) msg); |
|||
break; |
|||
case CLUSTER_EVENT_MSG: |
|||
processClusterEvent((ClusterEventMsg) msg); |
|||
break; |
|||
default: return false; |
|||
} |
|||
return true; |
|||
} |
|||
|
|||
private void processClusterEvent(ClusterEventMsg msg) { |
|||
processor.processClusterEvent(context(), msg); |
|||
} |
|||
|
|||
private void processTransportToSessionMsg(TransportToDeviceSessionActorMsg msg) { |
|||
initProcessor(msg); |
|||
processor.processToDeviceActorMsg(context(), msg); |
|||
} |
|||
|
|||
private void processActorsToSessionMsg(ActorSystemToDeviceSessionActorMsg msg) { |
|||
processor.processToDeviceMsg(context(), msg.getMsg()); |
|||
} |
|||
|
|||
private void processTimeoutMsg(SessionTimeoutMsg msg) { |
|||
if (processor != null) { |
|||
processor.processTimeoutMsg(context(), msg); |
|||
} else { |
|||
logger.warning("[{}] Can't process timeout msg: {} without processor", sessionId, msg); |
|||
} |
|||
} |
|||
|
|||
private void processSessionCloseMsg(SessionCtrlMsg msg) { |
|||
if (processor != null) { |
|||
processor.processSessionCtrlMsg(context(), msg); |
|||
} else if (msg instanceof SessionCloseMsg) { |
|||
AbstractSessionActorMsgProcessor.terminateSession(context(), sessionId); |
|||
} else { |
|||
logger.warning("[{}] Can't process session ctrl msg: {} without processor", sessionId, msg); |
|||
} |
|||
} |
|||
|
|||
private void initProcessor(TransportToDeviceSessionActorMsg msg) { |
|||
if (processor == null) { |
|||
SessionMsg sessionMsg = (SessionMsg) msg.getSessionMsg(); |
|||
if (sessionMsg.getSessionContext().getSessionType() == SessionType.SYNC) { |
|||
processor = new SyncMsgProcessor(systemContext, logger, sessionId); |
|||
} else { |
|||
processor = new ASyncMsgProcessor(systemContext, logger, sessionId); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public static class ActorCreator extends ContextBasedCreator<SessionActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final SessionId sessionId; |
|||
|
|||
public ActorCreator(ActorSystemContext context, SessionId sessionId) { |
|||
super(context); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public SessionActor create() throws Exception { |
|||
return new SessionActor(context, sessionId); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,180 +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.actors.session; |
|||
|
|||
import akka.actor.ActorInitializationException; |
|||
import akka.actor.ActorRef; |
|||
import akka.actor.InvalidActorNameException; |
|||
import akka.actor.LocalActorRef; |
|||
import akka.actor.OneForOneStrategy; |
|||
import akka.actor.Props; |
|||
import akka.actor.SupervisorStrategy; |
|||
import akka.actor.Terminated; |
|||
import akka.event.Logging; |
|||
import akka.event.LoggingAdapter; |
|||
import akka.japi.Function; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ContextAwareActor; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
import org.thingsboard.server.actors.service.DefaultActorService; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.SessionAwareMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.core.ActorSystemToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.core.SessionCloseMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionCtrlMsg; |
|||
import scala.concurrent.duration.Duration; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
|
|||
public class SessionManagerActor extends ContextAwareActor { |
|||
|
|||
private static final int INITIAL_SESSION_MAP_SIZE = 1024; |
|||
|
|||
private final LoggingAdapter log = Logging.getLogger(getContext().system(), this); |
|||
|
|||
private final Map<String, ActorRef> sessionActors; |
|||
|
|||
SessionManagerActor(ActorSystemContext systemContext) { |
|||
super(systemContext); |
|||
this.sessionActors = new HashMap<>(INITIAL_SESSION_MAP_SIZE); |
|||
} |
|||
|
|||
@Override |
|||
public SupervisorStrategy supervisorStrategy() { |
|||
return strategy; |
|||
} |
|||
|
|||
@Override |
|||
protected boolean process(TbActorMsg msg) { |
|||
//TODO Move everything here, to work with TbActorMsg
|
|||
return false; |
|||
} |
|||
|
|||
@Override |
|||
public void onReceive(Object msg) throws Exception { |
|||
if (msg instanceof SessionCtrlMsg) { |
|||
onSessionCtrlMsg((SessionCtrlMsg) msg); |
|||
} else if (msg instanceof SessionAwareMsg) { |
|||
forwardToSessionActor((SessionAwareMsg) msg); |
|||
} else if (msg instanceof SessionTerminationMsg) { |
|||
onSessionTermination((SessionTerminationMsg) msg); |
|||
} else if (msg instanceof Terminated) { |
|||
onTermination((Terminated) msg); |
|||
} else if (msg instanceof SessionTimeoutMsg) { |
|||
onSessionTimeout((SessionTimeoutMsg) msg); |
|||
} else if (msg instanceof ClusterEventMsg) { |
|||
broadcast(msg); |
|||
} |
|||
} |
|||
|
|||
private void broadcast(Object msg) { |
|||
sessionActors.values().forEach(actorRef -> actorRef.tell(msg, ActorRef.noSender())); |
|||
} |
|||
|
|||
private void onSessionTimeout(SessionTimeoutMsg msg) { |
|||
String sessionIdStr = msg.getSessionId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
sessionActor.tell(msg, ActorRef.noSender()); |
|||
} |
|||
} |
|||
|
|||
private void onSessionCtrlMsg(SessionCtrlMsg msg) { |
|||
String sessionIdStr = msg.getSessionId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
sessionActor.tell(msg, ActorRef.noSender()); |
|||
} |
|||
} |
|||
|
|||
private void onSessionTermination(SessionTerminationMsg msg) { |
|||
String sessionIdStr = msg.getId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.remove(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
log.debug("[{}] Removed session actor.", sessionIdStr); |
|||
//TODO: onSubscriptionUpdate device actor about session close;
|
|||
} else { |
|||
log.debug("[{}] Session actor was already removed.", sessionIdStr); |
|||
} |
|||
} |
|||
|
|||
private void forwardToSessionActor(SessionAwareMsg msg) { |
|||
if (msg instanceof ActorSystemToDeviceSessionActorMsg || msg instanceof SessionCloseMsg) { |
|||
String sessionIdStr = msg.getSessionId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
sessionActor.tell(msg, ActorRef.noSender()); |
|||
} else { |
|||
log.debug("[{}] Session actor was already removed.", sessionIdStr); |
|||
} |
|||
} else { |
|||
try { |
|||
getOrCreateSessionActor(msg.getSessionId()).tell(msg, self()); |
|||
} catch (InvalidActorNameException e) { |
|||
log.info("Invalid msg : {}", msg); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private ActorRef getOrCreateSessionActor(SessionId sessionId) { |
|||
String sessionIdStr = sessionId.toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor == null) { |
|||
log.debug("[{}] Creating session actor.", sessionIdStr); |
|||
sessionActor = context().actorOf( |
|||
Props.create(new SessionActor.ActorCreator(systemContext, sessionId)).withDispatcher(DefaultActorService.SESSION_DISPATCHER_NAME), |
|||
sessionIdStr); |
|||
sessionActors.put(sessionIdStr, sessionActor); |
|||
log.debug("[{}] Created session actor.", sessionIdStr); |
|||
} |
|||
return sessionActor; |
|||
} |
|||
|
|||
private void onTermination(Terminated message) { |
|||
ActorRef terminated = message.actor(); |
|||
if (terminated instanceof LocalActorRef) { |
|||
log.info("Removed actor: {}.", terminated); |
|||
//TODO: cleanup session actors map
|
|||
} else { |
|||
throw new IllegalStateException("Remote actors are not supported!"); |
|||
} |
|||
} |
|||
|
|||
public static class ActorCreator extends ContextBasedCreator<SessionManagerActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
public ActorCreator(ActorSystemContext context) { |
|||
super(context); |
|||
} |
|||
|
|||
@Override |
|||
public SessionManagerActor create() throws Exception { |
|||
return new SessionManagerActor(context); |
|||
} |
|||
} |
|||
|
|||
private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), new Function<Throwable, SupervisorStrategy.Directive>() { |
|||
@Override |
|||
public SupervisorStrategy.Directive apply(Throwable t) { |
|||
logger.error(t, "Unknown failure"); |
|||
return SupervisorStrategy.stop(); |
|||
} |
|||
}); |
|||
} |
|||
@ -1,95 +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.actors.session; |
|||
|
|||
import akka.actor.ActorContext; |
|||
import akka.event.LoggingAdapter; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.session.BasicSessionActorToAdaptorMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionContext; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg; |
|||
import org.thingsboard.server.common.msg.session.ex.SessionException; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
class SyncMsgProcessor extends AbstractSessionActorMsgProcessor { |
|||
private DeviceToDeviceActorMsg pendingMsg; |
|||
private Optional<ServerAddress> currentTargetServer; |
|||
private boolean pendingResponse; |
|||
|
|||
public SyncMsgProcessor(ActorSystemContext ctx, LoggingAdapter logger, SessionId sessionId) { |
|||
super(ctx, logger, sessionId); |
|||
} |
|||
|
|||
@Override |
|||
protected void processToDeviceActorMsg(ActorContext ctx, TransportToDeviceSessionActorMsg msg) { |
|||
updateSessionCtx(msg, SessionType.SYNC); |
|||
pendingMsg = toDeviceMsg(msg); |
|||
pendingResponse = true; |
|||
currentTargetServer = forwardToAppActor(ctx, pendingMsg); |
|||
scheduleMsgWithDelay(ctx, new SessionTimeoutMsg(sessionId), getTimeout(systemContext, msg.getSessionMsg().getSessionContext()), ctx.parent()); |
|||
} |
|||
|
|||
public void processTimeoutMsg(ActorContext context, SessionTimeoutMsg msg) { |
|||
if (pendingResponse) { |
|||
try { |
|||
sessionCtx.onMsg(SessionCloseMsg.onTimeout(sessionId)); |
|||
} catch (SessionException e) { |
|||
logger.warning("Failed to push session close msg", e); |
|||
} |
|||
terminateSession(context, this.sessionId); |
|||
} |
|||
} |
|||
|
|||
public void processToDeviceMsg(ActorContext context, ToDeviceMsg msg) { |
|||
try { |
|||
sessionCtx.onMsg(new BasicSessionActorToAdaptorMsg(this.sessionCtx, msg)); |
|||
pendingResponse = false; |
|||
} catch (SessionException e) { |
|||
logger.warning("Failed to push session response msg", e); |
|||
} |
|||
terminateSession(context, this.sessionId); |
|||
} |
|||
|
|||
@Override |
|||
public void processClusterEvent(ActorContext context, ClusterEventMsg msg) { |
|||
if (pendingResponse) { |
|||
Optional<ServerAddress> newTargetServer = forwardToAppActorIfAddressChanged(context, pendingMsg, currentTargetServer); |
|||
if (logger.isDebugEnabled()) { |
|||
if (!newTargetServer.equals(currentTargetServer)) { |
|||
if (newTargetServer.isPresent()) { |
|||
logger.debug("[{}] Forwarded msg to new server: {}", sessionId, newTargetServer.get()); |
|||
} else { |
|||
logger.debug("[{}] Forwarded msg to local server.", sessionId); |
|||
} |
|||
} |
|||
} |
|||
currentTargetServer = newTargetServer; |
|||
} |
|||
} |
|||
|
|||
private long getTimeout(ActorSystemContext ctx, SessionContext sessionCtx) { |
|||
return sessionCtx.getTimeout() > 0 ? sessionCtx.getTimeout() : ctx.getSyncSessionTimeout(); |
|||
} |
|||
} |
|||
@ -0,0 +1,336 @@ |
|||
/** |
|||
* 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.controller; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.http.HttpStatus; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.RequestBody; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestMethod; |
|||
import org.springframework.web.bind.annotation.RequestParam; |
|||
import org.springframework.web.bind.annotation.ResponseBody; |
|||
import org.springframework.web.bind.annotation.ResponseStatus; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.thingsboard.server.common.data.Customer; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.EntitySubtype; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.EntityView; |
|||
import org.thingsboard.server.common.data.audit.ActionType; |
|||
import org.thingsboard.server.common.data.entityview.EntityViewSearchQuery; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.EntityViewId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.UUIDBased; |
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
import org.thingsboard.server.common.data.page.TextPageData; |
|||
import org.thingsboard.server.common.data.page.TextPageLink; |
|||
import org.thingsboard.server.dao.exception.IncorrectParameterException; |
|||
import org.thingsboard.server.dao.model.ModelConstants; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
|
|||
import javax.annotation.Nullable; |
|||
import java.util.ArrayList; |
|||
import java.util.Collection; |
|||
import java.util.List; |
|||
import java.util.concurrent.ExecutionException; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.server.controller.CustomerController.CUSTOMER_ID; |
|||
|
|||
/** |
|||
* Created by Victor Basanets on 8/28/2017. |
|||
*/ |
|||
@RestController |
|||
@RequestMapping("/api") |
|||
@Slf4j |
|||
public class EntityViewController extends BaseController { |
|||
|
|||
public static final String ENTITY_VIEW_ID = "entityViewId"; |
|||
|
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
@RequestMapping(value = "/entityView/{entityViewId}", method = RequestMethod.GET) |
|||
@ResponseBody |
|||
public EntityView getEntityViewById(@PathVariable(ENTITY_VIEW_ID) String strEntityViewId) throws ThingsboardException { |
|||
checkParameter(ENTITY_VIEW_ID, strEntityViewId); |
|||
try { |
|||
return checkEntityViewId(new EntityViewId(toUUID(strEntityViewId))); |
|||
} catch (Exception e) { |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
@RequestMapping(value = "/entityView", method = RequestMethod.POST) |
|||
@ResponseBody |
|||
public EntityView saveEntityView(@RequestBody EntityView entityView) throws ThingsboardException { |
|||
try { |
|||
entityView.setTenantId(getCurrentUser().getTenantId()); |
|||
EntityView savedEntityView = checkNotNull(entityViewService.saveEntityView(entityView)); |
|||
List<ListenableFuture<List<Void>>> futures = new ArrayList<>(); |
|||
if (savedEntityView.getKeys() != null && savedEntityView.getKeys().getAttributes() != null) { |
|||
futures.add(copyAttributesFromEntityToEntityView(savedEntityView, DataConstants.CLIENT_SCOPE, savedEntityView.getKeys().getAttributes().getCs(), getCurrentUser())); |
|||
futures.add(copyAttributesFromEntityToEntityView(savedEntityView, DataConstants.SERVER_SCOPE, savedEntityView.getKeys().getAttributes().getSs(), getCurrentUser())); |
|||
futures.add(copyAttributesFromEntityToEntityView(savedEntityView, DataConstants.SHARED_SCOPE, savedEntityView.getKeys().getAttributes().getSh(), getCurrentUser())); |
|||
} |
|||
for (ListenableFuture<List<Void>> future : futures) { |
|||
try { |
|||
future.get(); |
|||
} catch (InterruptedException | ExecutionException e) { |
|||
throw new RuntimeException("Failed to copy attributes to entity view", e); |
|||
} |
|||
} |
|||
|
|||
logEntityAction(savedEntityView.getId(), savedEntityView, null, |
|||
entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); |
|||
return savedEntityView; |
|||
} catch (Exception e) { |
|||
logEntityAction(emptyId(EntityType.ENTITY_VIEW), entityView, null, |
|||
entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED, e); |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
private ListenableFuture<List<Void>> copyAttributesFromEntityToEntityView(EntityView entityView, String scope, Collection<String> keys, SecurityUser user) throws ThingsboardException { |
|||
EntityViewId entityId = entityView.getId(); |
|||
if (keys != null && !keys.isEmpty()) { |
|||
ListenableFuture<List<AttributeKvEntry>> getAttrFuture = attributesService.find(entityView.getEntityId(), scope, keys); |
|||
return Futures.transform(getAttrFuture, attributeKvEntries -> { |
|||
List<AttributeKvEntry> attributes; |
|||
if (attributeKvEntries != null && !attributeKvEntries.isEmpty()) { |
|||
attributes = |
|||
attributeKvEntries.stream() |
|||
.filter(attributeKvEntry -> { |
|||
long startTime = entityView.getStartTimeMs(); |
|||
long endTime = entityView.getEndTimeMs(); |
|||
long lastUpdateTs = attributeKvEntry.getLastUpdateTs(); |
|||
return startTime == 0 && endTime == 0 || |
|||
(endTime == 0 && startTime < lastUpdateTs) || |
|||
(startTime == 0 && endTime > lastUpdateTs) |
|||
? true : startTime < lastUpdateTs && endTime > lastUpdateTs; |
|||
}).collect(Collectors.toList()); |
|||
tsSubService.saveAndNotify(entityId, scope, attributes, new FutureCallback<Void>() { |
|||
@Override |
|||
public void onSuccess(@Nullable Void tmp) { |
|||
try { |
|||
logAttributesUpdated(user, entityId, scope, attributes, null); |
|||
} catch (ThingsboardException e) { |
|||
log.error("Failed to log attribute updates", e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
try { |
|||
logAttributesUpdated(user, entityId, scope, attributes, t); |
|||
} catch (ThingsboardException e) { |
|||
log.error("Failed to log attribute updates", e); |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
return null; |
|||
}); |
|||
} else { |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
} |
|||
|
|||
private void logAttributesUpdated(SecurityUser user, EntityId entityId, String scope, List<AttributeKvEntry> attributes, Throwable e) throws ThingsboardException { |
|||
logEntityAction(user, (UUIDBased & EntityId) entityId, null, null, ActionType.ATTRIBUTES_UPDATED, toException(e), |
|||
scope, attributes); |
|||
} |
|||
|
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@RequestMapping(value = "/entityView/{entityViewId}", method = RequestMethod.DELETE) |
|||
@ResponseStatus(value = HttpStatus.OK) |
|||
public void deleteEntityView(@PathVariable(ENTITY_VIEW_ID) String strEntityViewId) throws ThingsboardException { |
|||
checkParameter(ENTITY_VIEW_ID, strEntityViewId); |
|||
try { |
|||
EntityViewId entityViewId = new EntityViewId(toUUID(strEntityViewId)); |
|||
EntityView entityView = checkEntityViewId(entityViewId); |
|||
entityViewService.deleteEntityView(entityViewId); |
|||
logEntityAction(entityViewId, entityView, entityView.getCustomerId(), |
|||
ActionType.DELETED, null, strEntityViewId); |
|||
} catch (Exception e) { |
|||
logEntityAction(emptyId(EntityType.ENTITY_VIEW), |
|||
null, |
|||
null, |
|||
ActionType.DELETED, e, strEntityViewId); |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@RequestMapping(value = "/tenant/entityViews", params = {"entityViewName"}, method = RequestMethod.GET) |
|||
@ResponseBody |
|||
public EntityView getTenantEntityView( |
|||
@RequestParam String entityViewName) throws ThingsboardException { |
|||
try { |
|||
TenantId tenantId = getCurrentUser().getTenantId(); |
|||
return checkNotNull(entityViewService.findEntityViewByTenantIdAndName(tenantId, entityViewName)); |
|||
} catch (Exception e) { |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@RequestMapping(value = "/customer/{customerId}/entityView/{entityViewId}", method = RequestMethod.POST) |
|||
@ResponseBody |
|||
public EntityView assignEntityViewToCustomer(@PathVariable(CUSTOMER_ID) String strCustomerId, |
|||
@PathVariable(ENTITY_VIEW_ID) String strEntityViewId) throws ThingsboardException { |
|||
checkParameter(CUSTOMER_ID, strCustomerId); |
|||
checkParameter(ENTITY_VIEW_ID, strEntityViewId); |
|||
try { |
|||
CustomerId customerId = new CustomerId(toUUID(strCustomerId)); |
|||
Customer customer = checkCustomerId(customerId); |
|||
|
|||
EntityViewId entityViewId = new EntityViewId(toUUID(strEntityViewId)); |
|||
checkEntityViewId(entityViewId); |
|||
|
|||
EntityView savedEntityView = checkNotNull(entityViewService.assignEntityViewToCustomer(entityViewId, customerId)); |
|||
logEntityAction(entityViewId, savedEntityView, |
|||
savedEntityView.getCustomerId(), |
|||
ActionType.ASSIGNED_TO_CUSTOMER, null, strEntityViewId, strCustomerId, customer.getName()); |
|||
return savedEntityView; |
|||
} catch (Exception e) { |
|||
logEntityAction(emptyId(EntityType.ENTITY_VIEW), null, |
|||
null, |
|||
ActionType.ASSIGNED_TO_CUSTOMER, e, strEntityViewId, strCustomerId); |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@RequestMapping(value = "/customer/entityView/{entityViewId}", method = RequestMethod.DELETE) |
|||
@ResponseBody |
|||
public EntityView unassignEntityViewFromCustomer(@PathVariable(ENTITY_VIEW_ID) String strEntityViewId) throws ThingsboardException { |
|||
checkParameter(ENTITY_VIEW_ID, strEntityViewId); |
|||
try { |
|||
EntityViewId entityViewId = new EntityViewId(toUUID(strEntityViewId)); |
|||
EntityView entityView = checkEntityViewId(entityViewId); |
|||
if (entityView.getCustomerId() == null || entityView.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) { |
|||
throw new IncorrectParameterException("Entity View isn't assigned to any customer!"); |
|||
} |
|||
Customer customer = checkCustomerId(entityView.getCustomerId()); |
|||
EntityView savedEntityView = checkNotNull(entityViewService.unassignEntityViewFromCustomer(entityViewId)); |
|||
logEntityAction(entityViewId, entityView, |
|||
entityView.getCustomerId(), |
|||
ActionType.UNASSIGNED_FROM_CUSTOMER, null, strEntityViewId, customer.getId().toString(), customer.getName()); |
|||
|
|||
return savedEntityView; |
|||
} catch (Exception e) { |
|||
logEntityAction(emptyId(EntityType.ENTITY_VIEW), null, |
|||
null, |
|||
ActionType.UNASSIGNED_FROM_CUSTOMER, e, strEntityViewId); |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
@RequestMapping(value = "/customer/{customerId}/entityViews", params = {"limit"}, method = RequestMethod.GET) |
|||
@ResponseBody |
|||
public TextPageData<EntityView> getCustomerEntityViews( |
|||
@PathVariable("customerId") String strCustomerId, |
|||
@RequestParam int limit, |
|||
@RequestParam(required = false) String type, |
|||
@RequestParam(required = false) String textSearch, |
|||
@RequestParam(required = false) String idOffset, |
|||
@RequestParam(required = false) String textOffset) throws ThingsboardException { |
|||
checkParameter("customerId", strCustomerId); |
|||
try { |
|||
TenantId tenantId = getCurrentUser().getTenantId(); |
|||
CustomerId customerId = new CustomerId(toUUID(strCustomerId)); |
|||
checkCustomerId(customerId); |
|||
TextPageLink pageLink = createPageLink(limit, textSearch, idOffset, textOffset); |
|||
if (type != null && type.trim().length() > 0) { |
|||
return checkNotNull(entityViewService.findEntityViewsByTenantIdAndCustomerIdAndType(tenantId, customerId, pageLink, type)); |
|||
} else { |
|||
return checkNotNull(entityViewService.findEntityViewsByTenantIdAndCustomerId(tenantId, customerId, pageLink)); |
|||
} |
|||
} catch (Exception e) { |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@RequestMapping(value = "/tenant/entityViews", params = {"limit"}, method = RequestMethod.GET) |
|||
@ResponseBody |
|||
public TextPageData<EntityView> getTenantEntityViews( |
|||
@RequestParam int limit, |
|||
@RequestParam(required = false) String type, |
|||
@RequestParam(required = false) String textSearch, |
|||
@RequestParam(required = false) String idOffset, |
|||
@RequestParam(required = false) String textOffset) throws ThingsboardException { |
|||
try { |
|||
TenantId tenantId = getCurrentUser().getTenantId(); |
|||
TextPageLink pageLink = createPageLink(limit, textSearch, idOffset, textOffset); |
|||
|
|||
if (type != null && type.trim().length() > 0) { |
|||
return checkNotNull(entityViewService.findEntityViewByTenantIdAndType(tenantId, pageLink, type)); |
|||
} else { |
|||
return checkNotNull(entityViewService.findEntityViewByTenantId(tenantId, pageLink)); |
|||
} |
|||
} catch (Exception e) { |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
@RequestMapping(value = "/entityViews", method = RequestMethod.POST) |
|||
@ResponseBody |
|||
public List<EntityView> findByQuery(@RequestBody EntityViewSearchQuery query) throws ThingsboardException { |
|||
checkNotNull(query); |
|||
checkNotNull(query.getParameters()); |
|||
checkNotNull(query.getEntityViewTypes()); |
|||
checkEntityId(query.getParameters().getEntityId()); |
|||
try { |
|||
List<EntityView> entityViews = checkNotNull(entityViewService.findEntityViewsByQuery(query).get()); |
|||
entityViews = entityViews.stream().filter(entityView -> { |
|||
try { |
|||
checkEntityView(entityView); |
|||
return true; |
|||
} catch (ThingsboardException e) { |
|||
return false; |
|||
} |
|||
}).collect(Collectors.toList()); |
|||
return entityViews; |
|||
} catch (Exception e) { |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
@RequestMapping(value = "/entityView/types", method = RequestMethod.GET) |
|||
@ResponseBody |
|||
public List<EntitySubtype> getEntityViewTypes() throws ThingsboardException { |
|||
try { |
|||
SecurityUser user = getCurrentUser(); |
|||
TenantId tenantId = user.getTenantId(); |
|||
ListenableFuture<List<EntitySubtype>> entityViewTypes = entityViewService.findEntityViewTypesByTenantId(tenantId); |
|||
return checkNotNull(entityViewTypes.get()); |
|||
} catch (Exception e) { |
|||
throw handleException(e); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,63 @@ |
|||
/** |
|||
* 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.service.cluster.routing; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.service.cluster.discovery.ServerInstance; |
|||
|
|||
import java.util.concurrent.ConcurrentNavigableMap; |
|||
import java.util.concurrent.ConcurrentSkipListMap; |
|||
|
|||
/** |
|||
* Created by ashvayka on 23.09.18. |
|||
*/ |
|||
@Slf4j |
|||
public class ConsistentHashCircle { |
|||
private final ConcurrentNavigableMap<Long, ServerInstance> circle = |
|||
new ConcurrentSkipListMap<>(); |
|||
|
|||
public void put(long hash, ServerInstance instance) { |
|||
circle.put(hash, instance); |
|||
} |
|||
|
|||
public void remove(long hash) { |
|||
circle.remove(hash); |
|||
} |
|||
|
|||
public boolean isEmpty() { |
|||
return circle.isEmpty(); |
|||
} |
|||
|
|||
public boolean containsKey(Long hash) { |
|||
return circle.containsKey(hash); |
|||
} |
|||
|
|||
public ConcurrentNavigableMap<Long, ServerInstance> tailMap(Long hash) { |
|||
return circle.tailMap(hash); |
|||
} |
|||
|
|||
public Long firstKey() { |
|||
return circle.firstKey(); |
|||
} |
|||
|
|||
public ServerInstance get(Long hash) { |
|||
return circle.get(hash); |
|||
} |
|||
|
|||
public void log() { |
|||
circle.entrySet().forEach((e) -> log.debug("{} -> {}", e.getKey(), e.getValue().getServerAddress())); |
|||
} |
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
/** |
|||
* 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.service.install; |
|||
|
|||
import org.springframework.context.annotation.Profile; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.dao.util.NoSqlDao; |
|||
|
|||
@Service |
|||
@NoSqlDao |
|||
@Profile("install") |
|||
public class CassandraEntityDatabaseSchemaService extends CassandraAbstractDatabaseSchemaService |
|||
implements EntityDatabaseSchemaService { |
|||
public CassandraEntityDatabaseSchemaService() { |
|||
super("schema-entities.cql"); |
|||
} |
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
/** |
|||
* 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.service.install; |
|||
|
|||
import org.springframework.context.annotation.Profile; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.dao.util.NoSqlTsDao; |
|||
|
|||
@Service |
|||
@NoSqlTsDao |
|||
@Profile("install") |
|||
public class CassandraTsDatabaseSchemaService extends CassandraAbstractDatabaseSchemaService |
|||
implements TsDatabaseSchemaService { |
|||
public CassandraTsDatabaseSchemaService() { |
|||
super("schema-ts.cql"); |
|||
} |
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
/** |
|||
* 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.service.install; |
|||
|
|||
import org.springframework.context.annotation.Profile; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.dao.util.SqlDao; |
|||
|
|||
@Service |
|||
@SqlDao |
|||
@Profile("install") |
|||
public class SqlEntityDatabaseSchemaService extends SqlAbstractDatabaseSchemaService |
|||
implements EntityDatabaseSchemaService { |
|||
public SqlEntityDatabaseSchemaService() { |
|||
super("schema-entities.sql"); |
|||
} |
|||
} |
|||
@ -1,111 +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.service.queue; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.transport.quota.tenant.TenantQuotaService; |
|||
import org.thingsboard.server.dao.queue.MsgQueue; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicLong; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
public class DefaultMsgQueueService implements MsgQueueService { |
|||
|
|||
@Value("${actors.rule.queue.max_size}") |
|||
private long queueMaxSize; |
|||
|
|||
@Value("${actors.rule.queue.cleanup_period}") |
|||
private long queueCleanUpPeriod; |
|||
|
|||
@Autowired |
|||
private MsgQueue msgQueue; |
|||
|
|||
@Autowired |
|||
private TenantQuotaService quotaService; |
|||
|
|||
private ScheduledExecutorService cleanupExecutor; |
|||
|
|||
private Map<TenantId, AtomicLong> pendingCountPerTenant = new ConcurrentHashMap<>(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
if (queueCleanUpPeriod > 0) { |
|||
cleanupExecutor = Executors.newSingleThreadScheduledExecutor(); |
|||
cleanupExecutor.scheduleAtFixedRate(() -> cleanup(), |
|||
queueCleanUpPeriod, queueCleanUpPeriod, TimeUnit.SECONDS); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void stop() { |
|||
if (cleanupExecutor != null) { |
|||
cleanupExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> put(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) { |
|||
if(quotaService.isQuotaExceeded(tenantId.getId().toString())) { |
|||
log.warn("Tenant TbMsg Quota exceeded for [{}:{}] . Reject", tenantId.getId()); |
|||
return Futures.immediateFailedFuture(new RuntimeException("Tenant TbMsg Quota exceeded")); |
|||
} |
|||
|
|||
AtomicLong pendingMsgCount = pendingCountPerTenant.computeIfAbsent(tenantId, key -> new AtomicLong()); |
|||
if (pendingMsgCount.incrementAndGet() < queueMaxSize) { |
|||
return msgQueue.put(tenantId, msg, nodeId, clusterPartition); |
|||
} else { |
|||
pendingMsgCount.decrementAndGet(); |
|||
return Futures.immediateFailedFuture(new RuntimeException("Message queue is full!")); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> ack(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) { |
|||
ListenableFuture<Void> result = msgQueue.ack(tenantId, msg, nodeId, clusterPartition); |
|||
AtomicLong pendingMsgCount = pendingCountPerTenant.computeIfAbsent(tenantId, key -> new AtomicLong()); |
|||
pendingMsgCount.decrementAndGet(); |
|||
return result; |
|||
} |
|||
|
|||
@Override |
|||
public Iterable<TbMsg> findUnprocessed(TenantId tenantId, UUID nodeId, long clusterPartition) { |
|||
return msgQueue.findUnprocessed(tenantId, nodeId, clusterPartition); |
|||
} |
|||
|
|||
private void cleanup() { |
|||
pendingCountPerTenant.forEach((tenantId, pendingMsgCount) -> { |
|||
pendingMsgCount.set(0); |
|||
msgQueue.cleanUp(tenantId); |
|||
}); |
|||
} |
|||
|
|||
} |
|||
@ -1,32 +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.service.queue; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
public interface MsgQueueService { |
|||
|
|||
ListenableFuture<Void> put(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition); |
|||
|
|||
ListenableFuture<Void> ack(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition); |
|||
|
|||
Iterable<TbMsg> findUnprocessed(TenantId tenantId, UUID nodeId, long clusterPartition); |
|||
|
|||
} |
|||
@ -0,0 +1,102 @@ |
|||
/** |
|||
* 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.service.script; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
/** |
|||
* Created by ashvayka on 26.09.18. |
|||
*/ |
|||
@Slf4j |
|||
public abstract class AbstractJsInvokeService implements JsInvokeService { |
|||
|
|||
protected Map<UUID, String> scriptIdToNameMap = new ConcurrentHashMap<>(); |
|||
protected Map<UUID, AtomicInteger> blackListedFunctions = new ConcurrentHashMap<>(); |
|||
|
|||
@Override |
|||
public ListenableFuture<UUID> eval(JsScriptType scriptType, String scriptBody, String... argNames) { |
|||
UUID scriptId = UUID.randomUUID(); |
|||
String functionName = "invokeInternal_" + scriptId.toString().replace('-', '_'); |
|||
String jsScript = generateJsScript(scriptType, functionName, scriptBody, argNames); |
|||
return doEval(scriptId, functionName, jsScript); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Object> invokeFunction(UUID scriptId, Object... args) { |
|||
String functionName = scriptIdToNameMap.get(scriptId); |
|||
if (functionName == null) { |
|||
return Futures.immediateFailedFuture(new RuntimeException("No compiled script found for scriptId: [" + scriptId + "]!")); |
|||
} |
|||
if (!isBlackListed(scriptId)) { |
|||
return doInvokeFunction(scriptId, functionName, args); |
|||
} else { |
|||
return Futures.immediateFailedFuture( |
|||
new RuntimeException("Script is blacklisted due to maximum error count " + getMaxErrors() + "!")); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> release(UUID scriptId) { |
|||
String functionName = scriptIdToNameMap.get(scriptId); |
|||
if (functionName != null) { |
|||
try { |
|||
scriptIdToNameMap.remove(scriptId); |
|||
blackListedFunctions.remove(scriptId); |
|||
doRelease(scriptId, functionName); |
|||
} catch (Exception e) { |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
} |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
protected abstract ListenableFuture<UUID> doEval(UUID scriptId, String functionName, String scriptBody); |
|||
|
|||
protected abstract ListenableFuture<Object> doInvokeFunction(UUID scriptId, String functionName, Object[] args); |
|||
|
|||
protected abstract void doRelease(UUID scriptId, String functionName) throws Exception; |
|||
|
|||
protected abstract int getMaxErrors(); |
|||
|
|||
protected void onScriptExecutionError(UUID scriptId) { |
|||
blackListedFunctions.computeIfAbsent(scriptId, key -> new AtomicInteger(0)).incrementAndGet(); |
|||
} |
|||
|
|||
private String generateJsScript(JsScriptType scriptType, String functionName, String scriptBody, String... argNames) { |
|||
switch (scriptType) { |
|||
case RULE_NODE_SCRIPT: |
|||
return RuleNodeScriptFactory.generateRuleNodeScript(functionName, scriptBody, argNames); |
|||
default: |
|||
throw new RuntimeException("No script factory implemented for scriptType: " + scriptType); |
|||
} |
|||
} |
|||
|
|||
private boolean isBlackListed(UUID scriptId) { |
|||
if (blackListedFunctions.containsKey(scriptId)) { |
|||
AtomicInteger errorCount = blackListedFunctions.get(scriptId); |
|||
return errorCount.get() >= getMaxErrors(); |
|||
} else { |
|||
return false; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,109 @@ |
|||
/** |
|||
* 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.service.script; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import delight.nashornsandbox.NashornSandbox; |
|||
import delight.nashornsandbox.NashornSandboxes; |
|||
import jdk.nashorn.api.scripting.NashornScriptEngineFactory; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import javax.script.Invocable; |
|||
import javax.script.ScriptEngine; |
|||
import javax.script.ScriptException; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractNashornJsInvokeService extends AbstractJsInvokeService { |
|||
|
|||
private NashornSandbox sandbox; |
|||
private ScriptEngine engine; |
|||
private ExecutorService monitorExecutorService; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
if (useJsSandbox()) { |
|||
sandbox = NashornSandboxes.create(); |
|||
monitorExecutorService = Executors.newFixedThreadPool(getMonitorThreadPoolSize()); |
|||
sandbox.setExecutor(monitorExecutorService); |
|||
sandbox.setMaxCPUTime(getMaxCpuTime()); |
|||
sandbox.allowNoBraces(false); |
|||
sandbox.setMaxPreparedStatements(30); |
|||
} else { |
|||
NashornScriptEngineFactory factory = new NashornScriptEngineFactory(); |
|||
engine = factory.getScriptEngine(new String[]{"--no-java"}); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void stop() { |
|||
if (monitorExecutorService != null) { |
|||
monitorExecutorService.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
protected abstract boolean useJsSandbox(); |
|||
|
|||
protected abstract int getMonitorThreadPoolSize(); |
|||
|
|||
protected abstract long getMaxCpuTime(); |
|||
|
|||
@Override |
|||
protected ListenableFuture<UUID> doEval(UUID scriptId, String functionName, String jsScript) { |
|||
try { |
|||
if (useJsSandbox()) { |
|||
sandbox.eval(jsScript); |
|||
} else { |
|||
engine.eval(jsScript); |
|||
} |
|||
scriptIdToNameMap.put(scriptId, functionName); |
|||
} catch (Exception e) { |
|||
log.warn("Failed to compile JS script: {}", e.getMessage(), e); |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
return Futures.immediateFuture(scriptId); |
|||
} |
|||
|
|||
@Override |
|||
protected ListenableFuture<Object> doInvokeFunction(UUID scriptId, String functionName, Object[] args) { |
|||
try { |
|||
Object result; |
|||
if (useJsSandbox()) { |
|||
result = sandbox.getSandboxedInvocable().invokeFunction(functionName, args); |
|||
} else { |
|||
result = ((Invocable) engine).invokeFunction(functionName, args); |
|||
} |
|||
return Futures.immediateFuture(result); |
|||
} catch (Exception e) { |
|||
onScriptExecutionError(scriptId); |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
} |
|||
|
|||
protected void doRelease(UUID scriptId, String functionName) throws ScriptException { |
|||
if (useJsSandbox()) { |
|||
sandbox.eval(functionName + " = undefined;"); |
|||
} else { |
|||
engine.eval(functionName + " = undefined;"); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,188 +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.service.script; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import delight.nashornsandbox.NashornSandbox; |
|||
import delight.nashornsandbox.NashornSandboxes; |
|||
import jdk.nashorn.api.scripting.NashornScriptEngineFactory; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.tuple.Pair; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import javax.script.Invocable; |
|||
import javax.script.ScriptEngine; |
|||
import javax.script.ScriptException; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractNashornJsSandboxService implements JsSandboxService { |
|||
|
|||
private NashornSandbox sandbox; |
|||
private ScriptEngine engine; |
|||
private ExecutorService monitorExecutorService; |
|||
|
|||
private final Map<UUID, String> functionsMap = new ConcurrentHashMap<>(); |
|||
private final Map<UUID,AtomicInteger> blackListedFunctions = new ConcurrentHashMap<>(); |
|||
private final Map<String, Pair<UUID, AtomicInteger>> scriptToId = new ConcurrentHashMap<>(); |
|||
private final Map<UUID, AtomicInteger> scriptIdToCount = new ConcurrentHashMap<>(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
if (useJsSandbox()) { |
|||
sandbox = NashornSandboxes.create(); |
|||
monitorExecutorService = Executors.newFixedThreadPool(getMonitorThreadPoolSize()); |
|||
sandbox.setExecutor(monitorExecutorService); |
|||
sandbox.setMaxCPUTime(getMaxCpuTime()); |
|||
sandbox.allowNoBraces(false); |
|||
sandbox.setMaxPreparedStatements(30); |
|||
} else { |
|||
NashornScriptEngineFactory factory = new NashornScriptEngineFactory(); |
|||
engine = factory.getScriptEngine(new String[]{"--no-java"}); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void stop() { |
|||
if (monitorExecutorService != null) { |
|||
monitorExecutorService.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
protected abstract boolean useJsSandbox(); |
|||
|
|||
protected abstract int getMonitorThreadPoolSize(); |
|||
|
|||
protected abstract long getMaxCpuTime(); |
|||
|
|||
protected abstract int getMaxErrors(); |
|||
|
|||
@Override |
|||
public ListenableFuture<UUID> eval(JsScriptType scriptType, String scriptBody, String... argNames) { |
|||
Pair<UUID, AtomicInteger> deduplicated = deduplicate(scriptType, scriptBody); |
|||
UUID scriptId = deduplicated.getLeft(); |
|||
AtomicInteger duplicateCount = deduplicated.getRight(); |
|||
|
|||
if(duplicateCount.compareAndSet(0, 1)) { |
|||
String functionName = "invokeInternal_" + scriptId.toString().replace('-', '_'); |
|||
String jsScript = generateJsScript(scriptType, functionName, scriptBody, argNames); |
|||
try { |
|||
if (useJsSandbox()) { |
|||
sandbox.eval(jsScript); |
|||
} else { |
|||
engine.eval(jsScript); |
|||
} |
|||
functionsMap.put(scriptId, functionName); |
|||
} catch (Exception e) { |
|||
duplicateCount.decrementAndGet(); |
|||
log.warn("Failed to compile JS script: {}", e.getMessage(), e); |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
} else { |
|||
duplicateCount.incrementAndGet(); |
|||
} |
|||
return Futures.immediateFuture(scriptId); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Object> invokeFunction(UUID scriptId, Object... args) { |
|||
String functionName = functionsMap.get(scriptId); |
|||
if (functionName == null) { |
|||
return Futures.immediateFailedFuture(new RuntimeException("No compiled script found for scriptId: [" + scriptId + "]!")); |
|||
} |
|||
if (!isBlackListed(scriptId)) { |
|||
try { |
|||
Object result; |
|||
if (useJsSandbox()) { |
|||
result = sandbox.getSandboxedInvocable().invokeFunction(functionName, args); |
|||
} else { |
|||
result = ((Invocable)engine).invokeFunction(functionName, args); |
|||
} |
|||
return Futures.immediateFuture(result); |
|||
} catch (Exception e) { |
|||
blackListedFunctions.computeIfAbsent(scriptId, key -> new AtomicInteger(0)).incrementAndGet(); |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
} else { |
|||
return Futures.immediateFailedFuture( |
|||
new RuntimeException("Script is blacklisted due to maximum error count " + getMaxErrors() + "!")); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> release(UUID scriptId) { |
|||
AtomicInteger count = scriptIdToCount.get(scriptId); |
|||
if(count != null) { |
|||
if(count.decrementAndGet() > 0) { |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
} |
|||
|
|||
String functionName = functionsMap.get(scriptId); |
|||
if (functionName != null) { |
|||
try { |
|||
if (useJsSandbox()) { |
|||
sandbox.eval(functionName + " = undefined;"); |
|||
} else { |
|||
engine.eval(functionName + " = undefined;"); |
|||
} |
|||
functionsMap.remove(scriptId); |
|||
blackListedFunctions.remove(scriptId); |
|||
} catch (ScriptException e) { |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
} |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
private boolean isBlackListed(UUID scriptId) { |
|||
if (blackListedFunctions.containsKey(scriptId)) { |
|||
AtomicInteger errorCount = blackListedFunctions.get(scriptId); |
|||
return errorCount.get() >= getMaxErrors(); |
|||
} else { |
|||
return false; |
|||
} |
|||
} |
|||
|
|||
private String generateJsScript(JsScriptType scriptType, String functionName, String scriptBody, String... argNames) { |
|||
switch (scriptType) { |
|||
case RULE_NODE_SCRIPT: |
|||
return RuleNodeScriptFactory.generateRuleNodeScript(functionName, scriptBody, argNames); |
|||
default: |
|||
throw new RuntimeException("No script factory implemented for scriptType: " + scriptType); |
|||
} |
|||
} |
|||
|
|||
private Pair<UUID, AtomicInteger> deduplicate(JsScriptType scriptType, String scriptBody) { |
|||
Pair<UUID, AtomicInteger> precomputed = Pair.of(UUID.randomUUID(), new AtomicInteger()); |
|||
|
|||
Pair<UUID, AtomicInteger> pair = scriptToId.computeIfAbsent(deduplicateKey(scriptType, scriptBody), i -> precomputed); |
|||
AtomicInteger duplicateCount = scriptIdToCount.computeIfAbsent(pair.getLeft(), i -> pair.getRight()); |
|||
return Pair.of(pair.getLeft(), duplicateCount); |
|||
} |
|||
|
|||
private String deduplicateKey(JsScriptType scriptType, String scriptBody) { |
|||
return scriptType + "_" + scriptBody; |
|||
} |
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
/** |
|||
* 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.service.script; |
|||
|
|||
import java.util.List; |
|||
|
|||
/** |
|||
* Created by ashvayka on 25.09.18. |
|||
*/ |
|||
public class JsInvokeResponse { |
|||
|
|||
private String scriptId; |
|||
private String scriptBody; |
|||
private List<String> args; |
|||
|
|||
} |
|||
@ -0,0 +1,211 @@ |
|||
/** |
|||
* 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.service.script; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
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.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.gen.js.JsInvokeProtos; |
|||
import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; |
|||
import org.thingsboard.server.kafka.TBKafkaProducerTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaRequestTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.kafka.TbNodeIdProvider; |
|||
import org.thingsboard.server.service.cluster.discovery.DiscoveryService; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
|
|||
@Slf4j |
|||
@ConditionalOnProperty(prefix = "js", value = "evaluator", havingValue = "remote", matchIfMissing = true) |
|||
@Service |
|||
public class RemoteJsInvokeService extends AbstractJsInvokeService { |
|||
|
|||
@Autowired |
|||
private TbNodeIdProvider nodeIdProvider; |
|||
|
|||
@Autowired |
|||
private TbKafkaSettings kafkaSettings; |
|||
|
|||
@Value("${js.remote.request_topic}") |
|||
private String requestTopic; |
|||
|
|||
@Value("${js.remote.response_topic_prefix}") |
|||
private String responseTopicPrefix; |
|||
|
|||
@Value("${js.remote.max_pending_requests}") |
|||
private long maxPendingRequests; |
|||
|
|||
@Value("${js.remote.max_requests_timeout}") |
|||
private long maxRequestsTimeout; |
|||
|
|||
@Value("${js.remote.response_poll_interval}") |
|||
private int responsePollDuration; |
|||
|
|||
@Value("${js.remote.response_auto_commit_interval}") |
|||
private int autoCommitInterval; |
|||
|
|||
@Getter |
|||
@Value("${js.remote.max_errors}") |
|||
private int maxErrors; |
|||
|
|||
private TbKafkaRequestTemplate<JsInvokeProtos.RemoteJsRequest, JsInvokeProtos.RemoteJsResponse> kafkaTemplate; |
|||
private Map<UUID, String> scriptIdToBodysMap = new ConcurrentHashMap<>(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<JsInvokeProtos.RemoteJsRequest> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.defaultTopic(requestTopic); |
|||
requestBuilder.encoder(new RemoteJsRequestEncoder()); |
|||
requestBuilder.enricher((request, responseTopic, requestId) -> { |
|||
JsInvokeProtos.RemoteJsRequest.Builder remoteRequest = JsInvokeProtos.RemoteJsRequest.newBuilder(); |
|||
if (request.hasCompileRequest()) { |
|||
remoteRequest.setCompileRequest(request.getCompileRequest()); |
|||
} |
|||
if (request.hasInvokeRequest()) { |
|||
remoteRequest.setInvokeRequest(request.getInvokeRequest()); |
|||
} |
|||
if (request.hasReleaseRequest()) { |
|||
remoteRequest.setReleaseRequest(request.getReleaseRequest()); |
|||
} |
|||
remoteRequest.setResponseTopic(responseTopic); |
|||
remoteRequest.setRequestIdMSB(requestId.getMostSignificantBits()); |
|||
remoteRequest.setRequestIdLSB(requestId.getLeastSignificantBits()); |
|||
return remoteRequest.build(); |
|||
}); |
|||
|
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<JsInvokeProtos.RemoteJsResponse> responseBuilder = TBKafkaConsumerTemplate.builder(); |
|||
responseBuilder.settings(kafkaSettings); |
|||
responseBuilder.topic(responseTopicPrefix + "." + nodeIdProvider.getNodeId()); |
|||
responseBuilder.clientId("js-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.groupId("rule-engine-node-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.autoCommit(true); |
|||
responseBuilder.autoCommitIntervalMs(autoCommitInterval); |
|||
responseBuilder.decoder(new RemoteJsResponseDecoder()); |
|||
responseBuilder.requestIdExtractor((response) -> new UUID(response.getRequestIdMSB(), response.getRequestIdLSB())); |
|||
|
|||
TbKafkaRequestTemplate.TbKafkaRequestTemplateBuilder |
|||
<JsInvokeProtos.RemoteJsRequest, JsInvokeProtos.RemoteJsResponse> builder = TbKafkaRequestTemplate.builder(); |
|||
builder.requestTemplate(requestBuilder.build()); |
|||
builder.responseTemplate(responseBuilder.build()); |
|||
builder.maxPendingRequests(maxPendingRequests); |
|||
builder.maxRequestTimeout(maxRequestsTimeout); |
|||
builder.pollInterval(responsePollDuration); |
|||
kafkaTemplate = builder.build(); |
|||
kafkaTemplate.init(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
if (kafkaTemplate != null) { |
|||
kafkaTemplate.stop(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected ListenableFuture<UUID> doEval(UUID scriptId, String functionName, String scriptBody) { |
|||
JsInvokeProtos.JsCompileRequest jsRequest = JsInvokeProtos.JsCompileRequest.newBuilder() |
|||
.setScriptIdMSB(scriptId.getMostSignificantBits()) |
|||
.setScriptIdLSB(scriptId.getLeastSignificantBits()) |
|||
.setFunctionName(functionName) |
|||
.setScriptBody(scriptBody).build(); |
|||
|
|||
JsInvokeProtos.RemoteJsRequest jsRequestWrapper = JsInvokeProtos.RemoteJsRequest.newBuilder() |
|||
.setCompileRequest(jsRequest) |
|||
.build(); |
|||
|
|||
log.trace("Post compile request for scriptId [{}]", scriptId); |
|||
ListenableFuture<JsInvokeProtos.RemoteJsResponse> future = kafkaTemplate.post(scriptId.toString(), jsRequestWrapper); |
|||
return Futures.transform(future, response -> { |
|||
JsInvokeProtos.JsCompileResponse compilationResult = response.getCompileResponse(); |
|||
UUID compiledScriptId = new UUID(compilationResult.getScriptIdMSB(), compilationResult.getScriptIdLSB()); |
|||
if (compilationResult.getSuccess()) { |
|||
scriptIdToNameMap.put(scriptId, functionName); |
|||
scriptIdToBodysMap.put(scriptId, scriptBody); |
|||
return compiledScriptId; |
|||
} else { |
|||
log.debug("[{}] Failed to compile script due to [{}]: {}", compiledScriptId, compilationResult.getErrorCode().name(), compilationResult.getErrorDetails()); |
|||
throw new RuntimeException(compilationResult.getErrorDetails()); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
protected ListenableFuture<Object> doInvokeFunction(UUID scriptId, String functionName, Object[] args) { |
|||
String scriptBody = scriptIdToBodysMap.get(scriptId); |
|||
if (scriptBody == null) { |
|||
return Futures.immediateFailedFuture(new RuntimeException("No script body found for scriptId: [" + scriptId + "]!")); |
|||
} |
|||
JsInvokeProtos.JsInvokeRequest.Builder jsRequestBuilder = JsInvokeProtos.JsInvokeRequest.newBuilder() |
|||
.setScriptIdMSB(scriptId.getMostSignificantBits()) |
|||
.setScriptIdLSB(scriptId.getLeastSignificantBits()) |
|||
.setFunctionName(functionName) |
|||
.setTimeout((int) maxRequestsTimeout) |
|||
.setScriptBody(scriptIdToBodysMap.get(scriptId)); |
|||
|
|||
for (int i = 0; i < args.length; i++) { |
|||
jsRequestBuilder.addArgs(args[i].toString()); |
|||
} |
|||
|
|||
JsInvokeProtos.RemoteJsRequest jsRequestWrapper = JsInvokeProtos.RemoteJsRequest.newBuilder() |
|||
.setInvokeRequest(jsRequestBuilder.build()) |
|||
.build(); |
|||
|
|||
ListenableFuture<JsInvokeProtos.RemoteJsResponse> future = kafkaTemplate.post(scriptId.toString(), jsRequestWrapper); |
|||
return Futures.transform(future, response -> { |
|||
JsInvokeProtos.JsInvokeResponse invokeResult = response.getInvokeResponse(); |
|||
if (invokeResult.getSuccess()) { |
|||
return invokeResult.getResult(); |
|||
} else { |
|||
log.debug("[{}] Failed to compile script due to [{}]: {}", scriptId, invokeResult.getErrorCode().name(), invokeResult.getErrorDetails()); |
|||
throw new RuntimeException(invokeResult.getErrorDetails()); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
protected void doRelease(UUID scriptId, String functionName) throws Exception { |
|||
JsInvokeProtos.JsReleaseRequest jsRequest = JsInvokeProtos.JsReleaseRequest.newBuilder() |
|||
.setScriptIdMSB(scriptId.getMostSignificantBits()) |
|||
.setScriptIdLSB(scriptId.getLeastSignificantBits()) |
|||
.setFunctionName(functionName).build(); |
|||
|
|||
JsInvokeProtos.RemoteJsRequest jsRequestWrapper = JsInvokeProtos.RemoteJsRequest.newBuilder() |
|||
.setReleaseRequest(jsRequest) |
|||
.build(); |
|||
|
|||
ListenableFuture<JsInvokeProtos.RemoteJsResponse> future = kafkaTemplate.post(scriptId.toString(), jsRequestWrapper); |
|||
JsInvokeProtos.RemoteJsResponse response = future.get(); |
|||
|
|||
JsInvokeProtos.JsReleaseResponse compilationResult = response.getReleaseResponse(); |
|||
UUID compiledScriptId = new UUID(compilationResult.getScriptIdMSB(), compilationResult.getScriptIdLSB()); |
|||
if (compilationResult.getSuccess()) { |
|||
scriptIdToBodysMap.remove(scriptId); |
|||
} else { |
|||
log.debug("[{}] Failed to release script due", compiledScriptId); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* 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.service.script; |
|||
|
|||
import org.thingsboard.server.gen.js.JsInvokeProtos; |
|||
import org.thingsboard.server.kafka.TbKafkaDecoder; |
|||
|
|||
import java.io.IOException; |
|||
|
|||
/** |
|||
* Created by ashvayka on 25.09.18. |
|||
*/ |
|||
public class RemoteJsResponseDecoder implements TbKafkaDecoder<JsInvokeProtos.RemoteJsResponse> { |
|||
|
|||
@Override |
|||
public JsInvokeProtos.RemoteJsResponse decode(byte[] data) throws IOException { |
|||
return JsInvokeProtos.RemoteJsResponse.parseFrom(data); |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* 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.service.session; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.cache.annotation.CachePut; |
|||
import org.springframework.cache.annotation.Cacheable; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.DeviceSessionsCacheEntry; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Collections; |
|||
|
|||
import static org.thingsboard.server.common.data.CacheConstants.SESSIONS_CACHE; |
|||
|
|||
/** |
|||
* Created by ashvayka on 29.10.18. |
|||
*/ |
|||
@Service |
|||
@Slf4j |
|||
public class DefaultDeviceSessionCacheService implements DeviceSessionCacheService { |
|||
|
|||
@Override |
|||
@Cacheable(cacheNames = SESSIONS_CACHE, key = "#deviceId") |
|||
public DeviceSessionsCacheEntry get(DeviceId deviceId) { |
|||
log.debug("[{}] Fetching session data from cache", deviceId); |
|||
return DeviceSessionsCacheEntry.newBuilder().addAllSessions(Collections.emptyList()).build(); |
|||
} |
|||
|
|||
@Override |
|||
@CachePut(cacheNames = SESSIONS_CACHE, key = "#deviceId") |
|||
public DeviceSessionsCacheEntry put(DeviceId deviceId, DeviceSessionsCacheEntry sessions) { |
|||
log.debug("[{}] Pushing session data from cache: {}", deviceId, sessions); |
|||
return sessions; |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue