Igor Kulikov
5 years ago
19 changed files with 211 additions and 376 deletions
@ -1,91 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.event; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.hibernate.exception.ConstraintViolationException; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.jpa.repository.Modifying; |
|||
import org.springframework.transaction.PlatformTransactionManager; |
|||
import org.springframework.transaction.TransactionDefinition; |
|||
import org.springframework.transaction.TransactionStatus; |
|||
import org.springframework.transaction.support.DefaultTransactionDefinition; |
|||
import org.thingsboard.server.dao.model.sql.EventEntity; |
|||
|
|||
import javax.persistence.EntityManager; |
|||
import javax.persistence.PersistenceContext; |
|||
import javax.persistence.Query; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractEventInsertRepository implements EventInsertRepository { |
|||
|
|||
@PersistenceContext |
|||
protected EntityManager entityManager; |
|||
|
|||
@Autowired |
|||
protected PlatformTransactionManager transactionManager; |
|||
|
|||
protected EventEntity saveAndGet(EventEntity entity, String insertOrUpdateOnPrimaryKeyConflict, String insertOrUpdateOnUniqueKeyConflict) { |
|||
EventEntity eventEntity = null; |
|||
TransactionStatus insertTransaction = getTransactionStatus(TransactionDefinition.PROPAGATION_REQUIRED); |
|||
try { |
|||
eventEntity = processSaveOrUpdate(entity, insertOrUpdateOnPrimaryKeyConflict); |
|||
transactionManager.commit(insertTransaction); |
|||
} catch (Throwable throwable) { |
|||
transactionManager.rollback(insertTransaction); |
|||
if (throwable.getCause() instanceof ConstraintViolationException) { |
|||
log.trace("Insert request leaded in a violation of a defined integrity constraint {} for Entity with entityId {} and entityType {}", throwable.getMessage(), entity.getEventUid(), entity.getEventType()); |
|||
TransactionStatus transaction = getTransactionStatus(TransactionDefinition.PROPAGATION_REQUIRES_NEW); |
|||
try { |
|||
eventEntity = processSaveOrUpdate(entity, insertOrUpdateOnUniqueKeyConflict); |
|||
transactionManager.commit(transaction); |
|||
} catch (Throwable th) { |
|||
log.trace("Could not execute the update statement for Entity with entityId {} and entityType {}", entity.getEventUid(), entity.getEventType()); |
|||
transactionManager.rollback(transaction); |
|||
} |
|||
} else { |
|||
log.trace("Could not execute the insert statement for Entity with entityId {} and entityType {}", entity.getEventUid(), entity.getEventType()); |
|||
} |
|||
} |
|||
return eventEntity; |
|||
} |
|||
|
|||
@Modifying |
|||
protected abstract EventEntity doProcessSaveOrUpdate(EventEntity entity, String query); |
|||
|
|||
protected Query getQuery(EventEntity entity, String query) { |
|||
return entityManager.createNativeQuery(query, EventEntity.class) |
|||
.setParameter("id", entity.getUuid()) |
|||
.setParameter("created_time", entity.getCreatedTime()) |
|||
.setParameter("body", entity.getBody().toString()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("event_type", entity.getEventType()) |
|||
.setParameter("event_uid", entity.getEventUid()) |
|||
.setParameter("tenant_id", entity.getTenantId()) |
|||
.setParameter("ts", entity.getTs()); |
|||
} |
|||
|
|||
private EventEntity processSaveOrUpdate(EventEntity entity, String query) { |
|||
return doProcessSaveOrUpdate(entity, query); |
|||
} |
|||
|
|||
private TransactionStatus getTransactionStatus(int propagationRequired) { |
|||
DefaultTransactionDefinition insertDefinition = new DefaultTransactionDefinition(); |
|||
insertDefinition.setPropagationBehavior(propagationRequired); |
|||
return transactionManager.getTransaction(insertDefinition); |
|||
} |
|||
} |
|||
@ -1,63 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.event; |
|||
|
|||
import org.springframework.stereotype.Repository; |
|||
import org.thingsboard.server.dao.model.sql.EventEntity; |
|||
import org.thingsboard.server.dao.util.HsqlDao; |
|||
|
|||
import javax.persistence.Query; |
|||
|
|||
@HsqlDao |
|||
@Repository |
|||
public class HsqlEventInsertRepository extends AbstractEventInsertRepository { |
|||
|
|||
private static final String P_KEY_CONFLICT_STATEMENT = "(event.id=I.id)"; |
|||
private static final String UNQ_KEY_CONFLICT_STATEMENT = "(event.tenant_id=I.tenant_id AND event.entity_type=I.entity_type AND event.entity_id=I.entity_id AND event.event_type=I.event_type AND event.event_uid=I.event_uid)"; |
|||
|
|||
private static final String INSERT_OR_UPDATE_ON_P_KEY_CONFLICT = getInsertString(P_KEY_CONFLICT_STATEMENT); |
|||
private static final String INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT = getInsertString(UNQ_KEY_CONFLICT_STATEMENT); |
|||
|
|||
@Override |
|||
public EventEntity saveOrUpdate(EventEntity entity) { |
|||
return saveAndGet(entity, INSERT_OR_UPDATE_ON_P_KEY_CONFLICT, INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT); |
|||
} |
|||
|
|||
@Override |
|||
protected EventEntity doProcessSaveOrUpdate(EventEntity entity, String query) { |
|||
getQuery(entity, query).executeUpdate(); |
|||
return entityManager.find(EventEntity.class, entity.getUuid()); |
|||
} |
|||
|
|||
protected Query getQuery(EventEntity entity, String query) { |
|||
return entityManager.createNativeQuery(query, EventEntity.class) |
|||
.setParameter("id", entity.getUuid().toString()) |
|||
.setParameter("created_time", entity.getCreatedTime()) |
|||
.setParameter("body", entity.getBody().toString()) |
|||
.setParameter("entity_id", entity.getEntityId().toString()) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("event_type", entity.getEventType()) |
|||
.setParameter("event_uid", entity.getEventUid()) |
|||
.setParameter("tenant_id", entity.getTenantId().toString()) |
|||
.setParameter("ts", entity.getTs()); |
|||
} |
|||
|
|||
private static String getInsertString(String conflictStatement) { |
|||
return "MERGE INTO event USING (VALUES UUID(:id), :created_time, :body, UUID(:entity_id), :entity_type, :event_type, :event_uid, UUID(:tenant_id), :ts) I (id, created_time, body, entity_id, entity_type, event_type, event_uid, tenant_id, ts) ON " + conflictStatement |
|||
+ " WHEN MATCHED THEN UPDATE SET event.id = I.id, event.created_time = I.created_time, event.body = I.body, event.entity_id = I.entity_id, event.entity_type = I.entity_type, event.event_type = I.event_type, event.event_uid = I.event_uid, event.tenant_id = I.tenant_id, event.ts = I.ts" + |
|||
" WHEN NOT MATCHED THEN INSERT (id, created_time, body, entity_id, entity_type, event_type, event_uid, tenant_id, ts) VALUES (I.id, I.created_time, I.body, I.entity_id, I.entity_type, I.event_type, I.event_uid, I.tenant_id, I.ts)"; |
|||
} |
|||
} |
|||
@ -1,53 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.event; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.thingsboard.server.dao.model.sql.EventEntity; |
|||
import org.thingsboard.server.dao.util.PsqlDao; |
|||
|
|||
@Slf4j |
|||
@PsqlDao |
|||
@Repository |
|||
public class PsqlEventInsertRepository extends AbstractEventInsertRepository { |
|||
|
|||
private static final String P_KEY_CONFLICT_STATEMENT = "(id)"; |
|||
private static final String UNQ_KEY_CONFLICT_STATEMENT = "(tenant_id, created_time, entity_type, entity_id, event_type, event_uid)"; |
|||
|
|||
private static final String UPDATE_P_KEY_STATEMENT = "id = :id"; |
|||
private static final String UPDATE_UNQ_KEY_STATEMENT = "created_time = :created_time, tenant_id = :tenant_id, entity_type = :entity_type, entity_id = :entity_id, event_type = :event_type, event_uid = :event_uid"; |
|||
|
|||
private static final String INSERT_OR_UPDATE_ON_P_KEY_CONFLICT = getInsertOrUpdateString(P_KEY_CONFLICT_STATEMENT, UPDATE_UNQ_KEY_STATEMENT); |
|||
private static final String INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT = getInsertOrUpdateString(UNQ_KEY_CONFLICT_STATEMENT, UPDATE_P_KEY_STATEMENT); |
|||
|
|||
@Override |
|||
public EventEntity saveOrUpdate(EventEntity entity) { |
|||
return saveAndGet(entity, INSERT_OR_UPDATE_ON_P_KEY_CONFLICT, INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT); |
|||
} |
|||
|
|||
@Override |
|||
protected EventEntity doProcessSaveOrUpdate(EventEntity entity, String query) { |
|||
return (EventEntity) getQuery(entity, query).getSingleResult(); |
|||
|
|||
} |
|||
|
|||
private static String getInsertOrUpdateString(String eventKeyStatement, String updateKeyStatement) { |
|||
return "INSERT INTO event (id, created_time, body, entity_id, entity_type, event_type, event_uid, tenant_id, ts) " + |
|||
"VALUES (:id, :created_time, :body, :entity_id, :entity_type, :event_type, :event_uid, :tenant_id, :ts) " + |
|||
"ON CONFLICT " + eventKeyStatement + " DO UPDATE SET body = :body, ts = :ts," + updateKeyStatement + " returning *"; |
|||
} |
|||
} |
|||
Loading…
Reference in new issue