Volodymyr Babak
6 years ago
111 changed files with 1353 additions and 345 deletions
@ -0,0 +1,35 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.edge; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.TimePageData; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
|
|||
public interface EdgeEventService { |
|||
|
|||
EdgeEventType getEdgeEventTypeByEntityType(EntityType entityType); |
|||
|
|||
ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent); |
|||
|
|||
TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink); |
|||
|
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.common.data.edge; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.BaseData; |
|||
import org.thingsboard.server.common.data.id.EdgeEventId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
public class EdgeEvent extends BaseData<EdgeEventId> { |
|||
|
|||
private TenantId tenantId; |
|||
private EdgeId edgeId; |
|||
private String edgeEventAction; |
|||
private UUID entityId; |
|||
private EdgeEventType edgeEventType; |
|||
private transient JsonNode entityBody; |
|||
|
|||
public EdgeEvent() { |
|||
super(); |
|||
} |
|||
|
|||
public EdgeEvent(EdgeEventId id) { |
|||
super(id); |
|||
} |
|||
|
|||
public EdgeEvent(EdgeEvent event) { |
|||
super(event); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,35 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.common.data.id; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonCreator; |
|||
import com.fasterxml.jackson.annotation.JsonProperty; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
public class EdgeEventId extends UUIDBased { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
@JsonCreator |
|||
public EdgeEventId(@JsonProperty("id") UUID id) { |
|||
super(id); |
|||
} |
|||
|
|||
public static EdgeEventId fromString(String edgeEventId) { |
|||
return new EdgeEventId(UUID.fromString(edgeEventId)); |
|||
} |
|||
} |
|||
@ -0,0 +1,85 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.edge; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.TimePageData; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
import org.thingsboard.server.dao.exception.DataValidationException; |
|||
import org.thingsboard.server.dao.service.DataValidator; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
public class BaseEdgeEventService implements EdgeEventService { |
|||
|
|||
@Autowired |
|||
public EdgeEventDao edgeEventDao; |
|||
|
|||
@Override |
|||
public EdgeEventType getEdgeEventTypeByEntityType(EntityType entityType) { |
|||
switch (entityType) { |
|||
case DEVICE: |
|||
return EdgeEventType.DEVICE; |
|||
case ASSET: |
|||
return EdgeEventType.ASSET; |
|||
case ENTITY_VIEW: |
|||
return EdgeEventType.ENTITY_VIEW; |
|||
case DASHBOARD: |
|||
return EdgeEventType.DASHBOARD; |
|||
case USER: |
|||
return EdgeEventType.USER; |
|||
default: |
|||
log.warn("Failed to push notification to edge service. Unsupported entity type [{}]", entityType); |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent) { |
|||
edgeEventValidator.validate(edgeEvent, EdgeEvent::getTenantId); |
|||
return edgeEventDao.saveAsync(edgeEvent); |
|||
} |
|||
|
|||
@Override |
|||
public TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink) { |
|||
List<EdgeEvent> events = edgeEventDao.findEdgeEvents(tenantId.getId(), edgeId, pageLink); |
|||
return new TimePageData<>(events, pageLink); |
|||
} |
|||
|
|||
private DataValidator<EdgeEvent> edgeEventValidator = |
|||
new DataValidator<EdgeEvent>() { |
|||
@Override |
|||
protected void validateDataImpl(TenantId tenantId, EdgeEvent edgeEvent) { |
|||
if (edgeEvent.getEdgeId() == null) { |
|||
throw new DataValidationException("Edge id should be specified!"); |
|||
} |
|||
if (StringUtils.isEmpty(edgeEvent.getEdgeEventAction())) { |
|||
throw new DataValidationException("Edge Event action should be specified!"); |
|||
} |
|||
} |
|||
}; |
|||
} |
|||
@ -0,0 +1,51 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.edge; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
import org.thingsboard.server.dao.Dao; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* The Interface EdgeEventDao. |
|||
*/ |
|||
public interface EdgeEventDao extends Dao<EdgeEvent> { |
|||
|
|||
/** |
|||
* Save or update edge event object async |
|||
* |
|||
* @param edgeEvent the event object |
|||
* @return saved edge event object future |
|||
*/ |
|||
ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent); |
|||
|
|||
|
|||
/** |
|||
* Find edge events by tenantId, edgeId and pageLink. |
|||
* |
|||
* @param tenantId the tenantId |
|||
* @param edgeId the edgeId |
|||
* @param pageLink the pageLink |
|||
* @return the event list |
|||
*/ |
|||
List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink); |
|||
|
|||
} |
|||
@ -0,0 +1,135 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.model.nosql; |
|||
|
|||
import com.datastax.driver.core.utils.UUIDs; |
|||
import com.datastax.driver.mapping.annotations.ClusteringColumn; |
|||
import com.datastax.driver.mapping.annotations.Column; |
|||
import com.datastax.driver.mapping.annotations.PartitionKey; |
|||
import com.datastax.driver.mapping.annotations.Table; |
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.Event; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.EdgeEventId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.EntityIdFactory; |
|||
import org.thingsboard.server.common.data.id.EventId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.model.BaseEntity; |
|||
import org.thingsboard.server.dao.model.type.EdgeEventTypeCodec; |
|||
import org.thingsboard.server.dao.model.type.EntityTypeCodec; |
|||
import org.thingsboard.server.dao.model.type.JsonCodec; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_ACTION_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_COLUMN_FAMILY_NAME; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_EDGE_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_ENTITY_BODY_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_ENTITY_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_TENANT_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_TYPE_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_BODY_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_COLUMN_FAMILY_NAME; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_ENTITY_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_ENTITY_TYPE_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_TENANT_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_TYPE_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_UID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.ID_PROPERTY; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@Table(name = EDGE_EVENT_COLUMN_FAMILY_NAME) |
|||
public class EdgeEventEntity implements BaseEntity<EdgeEvent> { |
|||
|
|||
@Column(name = ID_PROPERTY) |
|||
private UUID id; |
|||
|
|||
@PartitionKey() |
|||
@Column(name = EDGE_EVENT_TENANT_ID_PROPERTY) |
|||
private UUID tenantId; |
|||
|
|||
@PartitionKey(value = 1) |
|||
@Column(name = EDGE_EVENT_EDGE_ID_PROPERTY) |
|||
private UUID edgeId; |
|||
|
|||
@PartitionKey(value = 2) |
|||
@Column(name = EDGE_EVENT_TYPE_PROPERTY, codec = EdgeEventTypeCodec.class) |
|||
private EdgeEventType edgeEventType; |
|||
|
|||
@PartitionKey(value = 3) |
|||
@Column(name = EDGE_EVENT_ENTITY_ID_PROPERTY) |
|||
private UUID entityId; |
|||
|
|||
@ClusteringColumn() |
|||
@Column(name = EDGE_EVENT_ACTION_PROPERTY) |
|||
private String edgeEventAction; |
|||
|
|||
// TODO
|
|||
@ClusteringColumn(value = 1) |
|||
@Column(name = EVENT_UID_PROPERTY) |
|||
private String eventUid; |
|||
|
|||
@Column(name = EDGE_EVENT_ENTITY_BODY_PROPERTY, codec = JsonCodec.class) |
|||
private JsonNode entityBody; |
|||
|
|||
public EdgeEventEntity(EdgeEvent edgeEvent) { |
|||
if (edgeEvent.getId() != null) { |
|||
this.id = edgeEvent.getId().getId(); |
|||
} |
|||
if (edgeEvent.getTenantId() != null) { |
|||
this.tenantId = edgeEvent.getTenantId().getId(); |
|||
} |
|||
if (edgeEvent.getEdgeId() != null) { |
|||
this.edgeId = edgeEvent.getEdgeId().getId(); |
|||
} |
|||
// if (event.getEntityId() != null) {
|
|||
// this.entityType = event.getEntityId().getEntityType();
|
|||
// this.entityId = event.getEntityId().getId();
|
|||
// }
|
|||
// this.edgeEventType = edgeEvent.getEdgeEventType();
|
|||
// this.edgeEventAction = edgeEvent.getEdgeEventAction();
|
|||
// this.entityBody = edgeEvent.getEntityBody();
|
|||
} |
|||
|
|||
@Override |
|||
public UUID getUuid() { |
|||
return id; |
|||
} |
|||
|
|||
@Override |
|||
public void setUuid(UUID id) { |
|||
this.id = id; |
|||
} |
|||
|
|||
@Override |
|||
public EdgeEvent toData() { |
|||
EdgeEvent edgeEvent = new EdgeEvent(new EdgeEventId(id)); |
|||
// edgeEvent.setCreatedTime(UUIDs.unixTimestamp(id));
|
|||
// edgeEvent.setTenantId(new TenantId(tenantId));
|
|||
// edgeEvent.setEdgeId(new EdgeId(edgeId));
|
|||
// edgeEvent.setEntityId(entityId);
|
|||
// event.setBody(body);
|
|||
// event.setType(eventType);
|
|||
// event.setUid(eventUid);
|
|||
return edgeEvent; |
|||
} |
|||
} |
|||
@ -0,0 +1,121 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.model.sql; |
|||
|
|||
import com.datastax.driver.core.utils.UUIDs; |
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import org.hibernate.annotations.Type; |
|||
import org.hibernate.annotations.TypeDef; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.EdgeEventId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.model.BaseEntity; |
|||
import org.thingsboard.server.dao.model.BaseSqlEntity; |
|||
import org.thingsboard.server.dao.util.mapping.JsonStringType; |
|||
|
|||
import javax.persistence.Column; |
|||
import javax.persistence.Entity; |
|||
import javax.persistence.EnumType; |
|||
import javax.persistence.Enumerated; |
|||
import javax.persistence.Table; |
|||
import java.util.UUID; |
|||
|
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_ACTION_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_COLUMN_FAMILY_NAME; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_EDGE_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_ENTITY_BODY_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_ENTITY_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_TENANT_ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EDGE_EVENT_TYPE_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.EPOCH_DIFF; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.TS_COLUMN; |
|||
|
|||
@Data |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@Entity |
|||
@TypeDef(name = "json", typeClass = JsonStringType.class) |
|||
@Table(name = EDGE_EVENT_COLUMN_FAMILY_NAME) |
|||
@NoArgsConstructor |
|||
public class EdgeEventEntity extends BaseSqlEntity<EdgeEvent> implements BaseEntity<EdgeEvent> { |
|||
|
|||
@Column(name = EDGE_EVENT_TENANT_ID_PROPERTY) |
|||
private String tenantId; |
|||
|
|||
@Column(name = EDGE_EVENT_EDGE_ID_PROPERTY) |
|||
private String edgeId; |
|||
|
|||
@Column(name = EDGE_EVENT_ENTITY_ID_PROPERTY) |
|||
private String entityId; |
|||
|
|||
@Enumerated(EnumType.STRING) |
|||
@Column(name = EDGE_EVENT_TYPE_PROPERTY) |
|||
private EdgeEventType edgeEventType; |
|||
|
|||
@Column(name = EDGE_EVENT_ACTION_PROPERTY) |
|||
private String edgeEventAction; |
|||
|
|||
@Type(type = "json") |
|||
@Column(name = EDGE_EVENT_ENTITY_BODY_PROPERTY) |
|||
private JsonNode entityBody; |
|||
|
|||
@Column(name = TS_COLUMN) |
|||
private long ts; |
|||
|
|||
public EdgeEventEntity(EdgeEvent edgeEvent) { |
|||
if (edgeEvent.getId() != null) { |
|||
this.setUuid(edgeEvent.getId().getId()); |
|||
this.ts = getTs(edgeEvent.getId().getId()); |
|||
} else { |
|||
this.ts = System.currentTimeMillis(); |
|||
} |
|||
if (edgeEvent.getTenantId() != null) { |
|||
this.tenantId = toString(edgeEvent.getTenantId().getId()); |
|||
} |
|||
if (edgeEvent.getEdgeId() != null) { |
|||
this.edgeId = toString(edgeEvent.getEdgeId().getId()); |
|||
} |
|||
if (edgeEvent.getEntityId() != null) { |
|||
this.entityId = toString(edgeEvent.getEntityId()); |
|||
} |
|||
this.edgeEventType = edgeEvent.getEdgeEventType(); |
|||
this.edgeEventAction = edgeEvent.getEdgeEventAction(); |
|||
this.entityBody = edgeEvent.getEntityBody(); |
|||
} |
|||
|
|||
@Override |
|||
public EdgeEvent toData() { |
|||
EdgeEvent edgeEvent = new EdgeEvent(new EdgeEventId(this.getUuid())); |
|||
edgeEvent.setCreatedTime(UUIDs.unixTimestamp(this.getUuid())); |
|||
edgeEvent.setTenantId(new TenantId(toUUID(tenantId))); |
|||
edgeEvent.setEdgeId(new EdgeId(toUUID(edgeId))); |
|||
if (entityId != null) { |
|||
edgeEvent.setEntityId(toUUID(entityId)); |
|||
} |
|||
edgeEvent.setEdgeEventType(edgeEventType); |
|||
edgeEvent.setEdgeEventAction(edgeEventAction); |
|||
edgeEvent.setEntityBody(entityBody); |
|||
return edgeEvent; |
|||
} |
|||
|
|||
private static long getTs(UUID uuid) { |
|||
return (uuid.timestamp() - EPOCH_DIFF) / 10000; |
|||
} |
|||
} |
|||
@ -0,0 +1,27 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.model.type; |
|||
|
|||
import com.datastax.driver.extras.codecs.enums.EnumNameCodec; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
|
|||
public class EdgeEventTypeCodec extends EnumNameCodec<EdgeEventType> { |
|||
|
|||
public EdgeEventTypeCodec() { |
|||
super(EdgeEventType.class); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.sql.edge; |
|||
|
|||
import org.springframework.data.jpa.repository.JpaSpecificationExecutor; |
|||
import org.springframework.data.repository.CrudRepository; |
|||
import org.thingsboard.server.dao.model.sql.EdgeEventEntity; |
|||
import org.thingsboard.server.dao.util.SqlDao; |
|||
|
|||
@SqlDao |
|||
public interface EdgeEventRepository extends CrudRepository<EdgeEventEntity, String>, JpaSpecificationExecutor<EdgeEventEntity> { |
|||
|
|||
} |
|||
@ -0,0 +1,112 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.sql.edge; |
|||
|
|||
import com.datastax.driver.core.utils.UUIDs; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.domain.PageRequest; |
|||
import org.springframework.data.domain.Pageable; |
|||
import org.springframework.data.domain.Sort; |
|||
import org.springframework.data.jpa.domain.Specification; |
|||
import org.springframework.data.repository.CrudRepository; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.UUIDConverter; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.id.EdgeEventId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
import org.thingsboard.server.dao.DaoUtil; |
|||
import org.thingsboard.server.dao.edge.EdgeEventDao; |
|||
import org.thingsboard.server.dao.model.sql.EdgeEventEntity; |
|||
import org.thingsboard.server.dao.sql.JpaAbstractSearchTimeDao; |
|||
import org.thingsboard.server.dao.util.SqlDao; |
|||
|
|||
import javax.persistence.criteria.Predicate; |
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
|
|||
import static org.thingsboard.server.dao.model.ModelConstants.ID_PROPERTY; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
@SqlDao |
|||
public class JpaBaseEdgeEventDao extends JpaAbstractSearchTimeDao<EdgeEventEntity, EdgeEvent> implements EdgeEventDao { |
|||
|
|||
private final UUID systemTenantId = NULL_UUID; |
|||
|
|||
@Autowired |
|||
private EdgeEventRepository edgeEventRepository; |
|||
|
|||
@Override |
|||
protected Class<EdgeEventEntity> getEntityClass() { |
|||
return EdgeEventEntity.class; |
|||
} |
|||
|
|||
@Override |
|||
protected CrudRepository<EdgeEventEntity, String> getCrudRepository() { |
|||
return edgeEventRepository; |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent) { |
|||
log.debug("Save edge event [{}] ", edgeEvent); |
|||
if (edgeEvent.getId() == null) { |
|||
edgeEvent.setId(new EdgeEventId(UUIDs.timeBased())); |
|||
} |
|||
return service.submit(() -> save(new EdgeEventEntity(edgeEvent)).orElse(null)); |
|||
} |
|||
|
|||
@Override |
|||
public List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink) { |
|||
Specification<EdgeEventEntity> timeSearchSpec = JpaAbstractSearchTimeDao.getTimeSearchPageSpec(pageLink, "id"); |
|||
Specification<EdgeEventEntity> fieldsSpec = getEntityFieldsSpec(tenantId, edgeId); |
|||
Sort.Direction sortDirection = pageLink.isAscOrder() ? Sort.Direction.ASC : Sort.Direction.DESC; |
|||
Pageable pageable = PageRequest.of(0, pageLink.getLimit(), sortDirection, ID_PROPERTY); |
|||
return DaoUtil.convertDataList(edgeEventRepository.findAll(Specification.where(timeSearchSpec).and(fieldsSpec), pageable).getContent()); |
|||
} |
|||
|
|||
public Optional<EdgeEvent> save(EdgeEventEntity entity) { |
|||
log.debug("Save edge event [{}] ", entity); |
|||
if (entity.getTenantId() == null) { |
|||
log.trace("Save system edge event with predefined id {}", systemTenantId); |
|||
entity.setTenantId(UUIDConverter.fromTimeUUID(systemTenantId)); |
|||
} |
|||
if (entity.getUuid() == null) { |
|||
entity.setUuid(UUIDs.timeBased()); |
|||
} |
|||
return Optional.of(DaoUtil.getData(edgeEventRepository.save(entity))); |
|||
} |
|||
|
|||
private Specification<EdgeEventEntity> getEntityFieldsSpec(UUID tenantId, EdgeId edgeId) { |
|||
return (root, criteriaQuery, criteriaBuilder) -> { |
|||
List<Predicate> predicates = new ArrayList<>(); |
|||
if (tenantId != null) { |
|||
Predicate tenantIdPredicate = criteriaBuilder.equal(root.get("tenantId"), UUIDConverter.fromTimeUUID(tenantId)); |
|||
predicates.add(tenantIdPredicate); |
|||
} |
|||
if (edgeId != null) { |
|||
Predicate entityIdPredicate = criteriaBuilder.equal(root.get("edgeId"), UUIDConverter.fromTimeUUID(edgeId.getId())); |
|||
predicates.add(entityIdPredicate); |
|||
} |
|||
return criteriaBuilder.and(predicates.toArray(new Predicate[]{})); |
|||
}; |
|||
} |
|||
} |
|||
@ -0,0 +1,108 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.service; |
|||
|
|||
import com.datastax.driver.core.utils.UUIDs; |
|||
import org.junit.Assert; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EdgeEventId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.TimePageData; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
|
|||
import java.io.IOException; |
|||
import java.time.LocalDateTime; |
|||
import java.time.Month; |
|||
import java.time.ZoneOffset; |
|||
|
|||
public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest { |
|||
|
|||
@Test |
|||
public void saveEdgeEvent() throws Exception { |
|||
EdgeId edgeId = new EdgeId(UUIDs.timeBased()); |
|||
DeviceId deviceId = new DeviceId(UUIDs.timeBased()); |
|||
EdgeEvent edgeEvent = generateEdgeEvent(null, edgeId, deviceId, DataConstants.ENTITY_CREATED); |
|||
EdgeEvent saved = edgeEventService.saveAsync(edgeEvent).get(); |
|||
Assert.assertEquals(saved.getTenantId(), edgeEvent.getTenantId()); |
|||
Assert.assertEquals(saved.getEdgeId(), edgeEvent.getEdgeId()); |
|||
Assert.assertEquals(saved.getEntityId(), edgeEvent.getEntityId()); |
|||
Assert.assertEquals(saved.getEdgeEventType(), edgeEvent.getEdgeEventType()); |
|||
Assert.assertEquals(saved.getEdgeEventAction(), edgeEvent.getEdgeEventAction()); |
|||
Assert.assertEquals(saved.getEntityBody(), edgeEvent.getEntityBody()); |
|||
} |
|||
|
|||
protected EdgeEvent generateEdgeEvent(TenantId tenantId, EdgeId edgeId, EntityId entityId, String edgeEventAction) throws IOException { |
|||
if (tenantId == null) { |
|||
tenantId = new TenantId(UUIDs.timeBased()); |
|||
} |
|||
EdgeEvent edgeEvent = new EdgeEvent(); |
|||
edgeEvent.setTenantId(tenantId); |
|||
edgeEvent.setEdgeId(edgeId); |
|||
edgeEvent.setEntityId(entityId.getId()); |
|||
edgeEvent.setEdgeEventType(EdgeEventType.DEVICE); |
|||
edgeEvent.setEdgeEventAction(edgeEventAction); |
|||
edgeEvent.setEntityBody(readFromResource("TestJsonData.json")); |
|||
return edgeEvent; |
|||
} |
|||
|
|||
|
|||
@Test |
|||
public void findEdgeEventsByTimeDescOrder() throws Exception { |
|||
long timeBeforeStartTime = LocalDateTime.of(2020, Month.NOVEMBER, 1, 11, 30).toEpochSecond(ZoneOffset.UTC); |
|||
long startTime = LocalDateTime.of(2020, Month.NOVEMBER, 1, 12, 0).toEpochSecond(ZoneOffset.UTC); |
|||
long eventTime = LocalDateTime.of(2020, Month.NOVEMBER, 1, 12, 30).toEpochSecond(ZoneOffset.UTC); |
|||
long endTime = LocalDateTime.of(2020, Month.NOVEMBER, 1, 13, 0).toEpochSecond(ZoneOffset.UTC); |
|||
long timeAfterEndTime = LocalDateTime.of(2020, Month.NOVEMBER, 1, 13, 30).toEpochSecond(ZoneOffset.UTC); |
|||
|
|||
EdgeId edgeId = new EdgeId(UUIDs.timeBased()); |
|||
DeviceId deviceId = new DeviceId(UUIDs.timeBased()); |
|||
TenantId tenantId = new TenantId(UUIDs.timeBased()); |
|||
saveEdgeEventWithProvidedTime(timeBeforeStartTime, edgeId, deviceId, tenantId); |
|||
EdgeEvent savedEdgeEvent = saveEdgeEventWithProvidedTime(eventTime, edgeId, deviceId, tenantId); |
|||
EdgeEvent savedEdgeEvent2 = saveEdgeEventWithProvidedTime(eventTime + 1, edgeId, deviceId, tenantId); |
|||
EdgeEvent savedEdgeEvent3 = saveEdgeEventWithProvidedTime(eventTime + 2, edgeId, deviceId, tenantId); |
|||
saveEdgeEventWithProvidedTime(timeAfterEndTime, edgeId, deviceId, tenantId); |
|||
|
|||
TimePageData<EdgeEvent> edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, new TimePageLink(2, startTime, endTime, false)); |
|||
|
|||
Assert.assertNotNull(edgeEvents.getData()); |
|||
Assert.assertTrue(edgeEvents.getData().size() == 2); |
|||
Assert.assertTrue(edgeEvents.getData().get(0).getUuidId().equals(savedEdgeEvent3.getUuidId())); |
|||
Assert.assertTrue(edgeEvents.getData().get(1).getUuidId().equals(savedEdgeEvent2.getUuidId())); |
|||
Assert.assertTrue(edgeEvents.hasNext()); |
|||
Assert.assertNotNull(edgeEvents.getNextPageLink()); |
|||
|
|||
edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, edgeEvents.getNextPageLink()); |
|||
|
|||
Assert.assertNotNull(edgeEvents.getData()); |
|||
Assert.assertTrue(edgeEvents.getData().size() == 1); |
|||
Assert.assertTrue(edgeEvents.getData().get(0).getUuidId().equals(savedEdgeEvent.getUuidId())); |
|||
Assert.assertFalse(edgeEvents.hasNext()); |
|||
Assert.assertNull(edgeEvents.getNextPageLink()); |
|||
} |
|||
|
|||
private EdgeEvent saveEdgeEventWithProvidedTime(long time, EdgeId edgeId, EntityId entityId, TenantId tenantId) throws Exception { |
|||
EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, entityId, DataConstants.ENTITY_CREATED); |
|||
edgeEvent.setId(new EdgeEventId(UUIDs.startOf(time))); |
|||
return edgeEventService.saveAsync(edgeEvent).get(); |
|||
} |
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.dao.service.nosql; |
|||
|
|||
import org.thingsboard.server.dao.service.BaseEdgeEventServiceTest; |
|||
import org.thingsboard.server.dao.service.DaoNoSqlTest; |
|||
|
|||
@DaoNoSqlTest |
|||
public class EdgeEventServiceNoSqlTest extends BaseEdgeEventServiceTest { |
|||
} |
|||
@ -1,41 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.rule.engine.edge; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import lombok.Data; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import javax.annotation.Nullable; |
|||
|
|||
import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS; |
|||
|
|||
@Data |
|||
class PushToEdgeNodeCallback implements FutureCallback<Void> { |
|||
private final TbContext ctx; |
|||
private final TbMsg msg; |
|||
|
|||
@Override |
|||
public void onSuccess(@Nullable Void result) { |
|||
ctx.tellNext(msg, SUCCESS); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
ctx.tellFailure(msg, t); |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue