@ -29,10 +29,12 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory ;
import org.thingsboard.server.common.data.id.RuleChainId ;
import org.thingsboard.server.common.data.id.RuleNodeId ;
import org.thingsboard.server.common.data.msg.TbMsgType ;
import org.thingsboard.server.common.msg.gen.MsgProtos ;
import org.thingsboard.server.common.msg.queue.TbMsgCallback ;
import java.io.Serializable ;
import java.util.Objects ;
import java.util.UUID ;
/ * *
@ -42,6 +44,10 @@ import java.util.UUID;
@Slf4j
public final class TbMsg implements Serializable {
public static final String EMPTY_JSON_OBJECT = "{}" ;
public static final String EMPTY_JSON_ARRAY = "[]" ;
public static final String EMPTY_STRING = "" ;
private final String queueName ;
private final UUID id ;
private final long ts ;
@ -66,82 +72,234 @@ public final class TbMsg implements Serializable {
return ctx . getAndIncrementRuleNodeCounter ( ) ;
}
@Deprecated ( since = "3.5.2" , forRemoval = true )
public static TbMsg newMsg ( String queueName , String type , EntityId originator , TbMsgMetaData metaData , String data , RuleChainId ruleChainId , RuleNodeId ruleNodeId ) {
return newMsg ( queueName , type , originator , null , metaData , data , ruleChainId , ruleNodeId ) ;
}
/ * *
* Creates a new TbMsg instance with the specified parameters .
*
* < p > < strong > Deprecated : < / strong > This method is deprecated since version 3 . 5 . 2 and should only be used when you need to
* specify a custom message type that doesn ' t exist in the { @link TbMsgType } enum . For standard message types ,
* it is recommended to use the { @link # newMsg ( String , TbMsgType , EntityId , CustomerId , TbMsgMetaData , String , RuleChainId , RuleNodeId ) }
* method instead . < / p >
*
* @param queueName the name of the queue where the message will be sent
* @param type the type of the message
* @param originator the originator of the message
* @param customerId the ID of the customer associated with the message
* @param metaData the metadata of the message
* @param data the data of the message
* @param ruleChainId the ID of the rule chain associated with the message
* @param ruleNodeId the ID of the rule node associated with the message
* @return new TbMsg instance
* /
@Deprecated ( since = "3.5.2" )
public static TbMsg newMsg ( String queueName , String type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , String data , RuleChainId ruleChainId , RuleNodeId ruleNodeId ) {
return new TbMsg ( queueName , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type , originator , customerId ,
metaData . copy ( ) , TbMsgDataType . JSON , data , ruleChainId , ruleNodeId , null , TbMsgCallback . EMPTY ) ;
}
@Deprecated ( since = "3.5.2" , forRemoval = true )
public static TbMsg newMsg ( String type , EntityId originator , TbMsgMetaData metaData , String data ) {
return newMsg ( type , originator , null , metaData , data ) ;
}
@Deprecated ( since = "3.5.2" , forRemoval = true )
public static TbMsg newMsg ( String type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , String data ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type , originator , customerId ,
metaData . copy ( ) , TbMsgDataType . JSON , data , null , null , null , TbMsgCallback . EMPTY ) ;
}
public static TbMsg newMsg ( String queueName , TbMsgType type , EntityId originator , TbMsgMetaData metaData , String data , RuleChainId ruleChainId , RuleNodeId ruleNodeId ) {
return newMsg ( queueName , type , originator , null , metaData , data , ruleChainId , ruleNodeId ) ;
}
public static TbMsg newMsg ( String queueName , TbMsgType type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , String data , RuleChainId ruleChainId , RuleNodeId ruleNodeId ) {
return new TbMsg ( queueName , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type . name ( ) , originator , customerId ,
metaData . copy ( ) , TbMsgDataType . JSON , data , ruleChainId , ruleNodeId , null , TbMsgCallback . EMPTY ) ;
}
public static TbMsg newMsg ( TbMsgType type , EntityId originator , TbMsgMetaData metaData , String data ) {
return newMsg ( type , originator , null , metaData , data ) ;
}
public static TbMsg newMsg ( TbMsgType type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , String data ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type . name ( ) , originator , customerId ,
metaData . copy ( ) , TbMsgDataType . JSON , data , null , null , null , TbMsgCallback . EMPTY ) ;
}
// REALLY NEW MSG
/ * *
* Creates a new TbMsg instance with the specified parameters .
*
* < p > < strong > Deprecated : < / strong > This method is deprecated since version 3 . 5 . 2 and should only be used when you need to
* specify a custom message type that doesn ' t exist in the { @link TbMsgType } enum . For standard message types ,
* it is recommended to use the { @link # newMsg ( String , TbMsgType , EntityId , TbMsgMetaData , String ) }
* method instead . < / p >
*
* @param queueName the name of the queue where the message will be sent
* @param type the type of the message
* @param originator the originator of the message
* @param metaData the metadata of the message
* @param data the data of the message
* @return new TbMsg instance
* /
@Deprecated ( since = "3.5.2" )
public static TbMsg newMsg ( String queueName , String type , EntityId originator , TbMsgMetaData metaData , String data ) {
return newMsg ( queueName , type , originator , null , metaData , data ) ;
}
/ * *
* Creates a new TbMsg instance with the specified parameters .
*
* < p > < strong > Deprecated : < / strong > This method is deprecated since version 3 . 5 . 2 and should only be used when you need to
* specify a custom message type that doesn ' t exist in the { @link TbMsgType } enum . For standard message types ,
* it is recommended to use the { @link # newMsg ( String , TbMsgType , EntityId , CustomerId , TbMsgMetaData , String ) }
* method instead . < / p >
*
* @param queueName the name of the queue where the message will be sent
* @param type the type of the message
* @param originator the originator of the message
* @param customerId the ID of the customer associated with the message
* @param metaData the metadata of the message
* @param data the data of the message
* @return new TbMsg instance
* /
@Deprecated ( since = "3.5.2" )
public static TbMsg newMsg ( String queueName , String type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , String data ) {
return new TbMsg ( queueName , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type , originator , customerId ,
metaData . copy ( ) , TbMsgDataType . JSON , data , null , null , null , TbMsgCallback . EMPTY ) ;
}
@Deprecated ( since = "3.5.2" , forRemoval = true )
public static TbMsg newMsg ( String type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , TbMsgDataType dataType , String data ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type , originator , customerId ,
metaData . copy ( ) , dataType , data , null , null , null , TbMsgCallback . EMPTY ) ;
}
/ * *
* Creates a new TbMsg instance with the specified parameters .
*
* < p > < strong > Deprecated : < / strong > This method is deprecated since version 3 . 5 . 2 and should only be used when you need to
* specify a custom message type that doesn ' t exist in the { @link TbMsgType } enum . For standard message types ,
* it is recommended to use the { @link # newMsg ( TbMsgType , EntityId , TbMsgMetaData , TbMsgDataType , String ) }
* method instead . < / p >
*
* @param type the type of the message
* @param originator the originator of the message
* @param metaData the metadata of the message
* @param dataType the dataType of the message
* @param data the data of the message
* @return new TbMsg instance
* /
@Deprecated ( since = "3.5.2" )
public static TbMsg newMsg ( String type , EntityId originator , TbMsgMetaData metaData , TbMsgDataType dataType , String data ) {
return newMsg ( type , originator , null , metaData , dataType , data ) ;
}
public static TbMsg newMsg ( String queueName , TbMsgType type , EntityId originator , TbMsgMetaData metaData , String data ) {
return newMsg ( queueName , type , originator , null , metaData , data ) ;
}
public static TbMsg newMsg ( String queueName , TbMsgType type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , String data ) {
return new TbMsg ( queueName , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type . name ( ) , originator , customerId ,
metaData . copy ( ) , TbMsgDataType . JSON , data , null , null , null , TbMsgCallback . EMPTY ) ;
}
public static TbMsg newMsg ( TbMsgType type , EntityId originator , CustomerId customerId , TbMsgMetaData metaData , TbMsgDataType dataType , String data ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type . name ( ) , originator , customerId ,
metaData . copy ( ) , dataType , data , null , null , null , TbMsgCallback . EMPTY ) ;
}
public static TbMsg newMsg ( TbMsgType type , EntityId originator , TbMsgMetaData metaData , TbMsgDataType dataType , String data ) {
return newMsg ( type , originator , null , metaData , dataType , data ) ;
}
// For Tests only
@Deprecated ( since = "3.5.2" , forRemoval = true )
public static TbMsg newMsg ( String type , EntityId originator , TbMsgMetaData metaData , TbMsgDataType dataType , String data , RuleChainId ruleChainId , RuleNodeId ruleNodeId ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type , originator , null ,
metaData . copy ( ) , dataType , data , ruleChainId , ruleNodeId , null , TbMsgCallback . EMPTY ) ;
}
@Deprecated ( since = "3.5.2" , forRemoval = true )
public static TbMsg newMsg ( String type , EntityId originator , TbMsgMetaData metaData , String data , TbMsgCallback callback ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type , originator , null ,
metaData . copy ( ) , TbMsgDataType . JSON , data , null , null , null , callback ) ;
}
/ * *
* Transforms an existing TbMsg instance by changing its message type , originator , metadata , and data .
*
* < p > < strong > Deprecated : < / strong > This method is deprecated since version 3 . 5 . 2 and should only be used when you need to
* specify a custom message type that doesn ' t exist in the { @link TbMsgType } enum . For standard message types ,
* it is recommended to use the { @link # transformMsg ( TbMsg , TbMsgType , EntityId , TbMsgMetaData , String ) }
* method instead . < / p >
*
*
* @param tbMsg the TbMsg instance to transform
* @param type the new message type
* @param originator the new originator
* @param metaData the new metadata
* @param data the new data
* @return the transformed TbMsg instance
* /
@Deprecated ( since = "3.5.2" )
public static TbMsg transformMsg ( TbMsg tbMsg , String type , EntityId originator , TbMsgMetaData metaData , String data ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , type , originator , tbMsg . customerId , metaData . copy ( ) , tbMsg . dataType ,
data , tbMsg . ruleChainId , tbMsg . ruleNodeId , tbMsg . ctx . copy ( ) , tbMsg . callback ) ;
}
public static TbMsg newMsg ( TbMsgType type , EntityId originator , TbMsgMetaData metaData , TbMsgDataType dataType , String data , RuleChainId ruleChainId , RuleNodeId ruleNodeId ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type . name ( ) , originator , null ,
metaData . copy ( ) , dataType , data , ruleChainId , ruleNodeId , null , TbMsgCallback . EMPTY ) ;
}
public static TbMsg newMsg ( TbMsgType type , EntityId originator , TbMsgMetaData metaData , String data , TbMsgCallback callback ) {
return new TbMsg ( null , UUID . randomUUID ( ) , System . currentTimeMillis ( ) , type . name ( ) , originator , null ,
metaData . copy ( ) , TbMsgDataType . JSON , data , null , null , null , callback ) ;
}
public static TbMsg transformMsg ( TbMsg tbMsg , TbMsgType type , EntityId originator , TbMsgMetaData metaData , String data ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , type . name ( ) , originator , tbMsg . customerId , metaData . copy ( ) , tbMsg . dataType ,
data , tbMsg . ruleChainId , tbMsg . ruleNodeId , tbMsg . ctx . copy ( ) , tbMsg . callback ) ;
}
public static TbMsg transformMsgOriginator ( TbMsg tbMsg , EntityId originatorId ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , tbMsg . type , originatorId , tbMsg . getCustomerId ( ) , tbMsg . metaData , tbMsg . dataType ,
tbMsg . data , tbMsg . ruleChainId , tbMsg . ruleNodeId , tbMsg . ctx . copy ( ) , tbMsg . getCallback ( ) ) ;
}
public static TbMsg transformMsgData ( TbMsg tbMsg , String data ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , tbMsg . type , tbMsg . originator , tbMsg . customerId , tbMsg . metaData , tbMsg . dataType ,
data , tbMsg . ruleChainId , tbMsg . ruleNodeId , tbMsg . ctx . copy ( ) , tbMsg . getCallback ( ) ) ;
}
public static TbMsg transformMsg ( TbMsg tbMsg , TbMsgMetaData metadata ) {
public static TbMsg transformMsgMetadata ( TbMsg tbMsg , TbMsgMetaData metadata ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , tbMsg . type , tbMsg . originator , tbMsg . customerId , metadata . copy ( ) , tbMsg . dataType ,
tbMsg . data , tbMsg . ruleChainId , tbMsg . ruleNodeId , tbMsg . ctx . copy ( ) , tbMsg . getCallback ( ) ) ;
}
public static TbMsg transformMsg ( TbMsg tbMsg , CustomerId customerId ) {
public static TbMsg transformMsg ( TbMsg tbMsg , TbMsgMetaData metadata , String data ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , tbMsg . type , tbMsg . originator , tbMsg . customerId , metadata , tbMsg . dataType ,
data , tbMsg . ruleChainId , tbMsg . ruleNodeId , tbMsg . ctx . copy ( ) , tbMsg . getCallback ( ) ) ;
}
public static TbMsg transformMsgCustomerId ( TbMsg tbMsg , CustomerId customerId ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , tbMsg . type , tbMsg . originator , customerId , tbMsg . metaData , tbMsg . dataType ,
tbMsg . data , tbMsg . ruleChainId , tbMsg . ruleNodeId , tbMsg . ctx . copy ( ) , tbMsg . getCallback ( ) ) ;
}
public static TbMsg transformMsg ( TbMsg tbMsg , RuleChainId ruleChainId ) {
public static TbMsg transformMsgRuleChainId ( TbMsg tbMsg , RuleChainId ruleChainId ) {
return new TbMsg ( tbMsg . queueName , tbMsg . id , tbMsg . ts , tbMsg . type , tbMsg . originator , tbMsg . customerId , tbMsg . metaData , tbMsg . dataType ,
tbMsg . data , ruleChainId , null , tbMsg . ctx . copy ( ) , tbMsg . getCallback ( ) ) ;
}
public static TbMsg transformMsg ( TbMsg tbMsg , String queueName ) {
public static TbMsg transformMsgQueueName ( TbMsg tbMsg , String queueName ) {
return new TbMsg ( queueName , tbMsg . id , tbMsg . ts , tbMsg . type , tbMsg . originator , tbMsg . customerId , tbMsg . metaData , tbMsg . dataType ,
tbMsg . data , tbMsg . getRuleChainId ( ) , null , tbMsg . ctx . copy ( ) , tbMsg . getCallback ( ) ) ;
}
@ -183,11 +341,7 @@ public final class TbMsg implements Serializable {
this . ruleChainId = ruleChainId ;
this . ruleNodeId = ruleNodeId ;
this . ctx = ctx ! = null ? ctx : new TbMsgProcessingCtx ( ) ;
if ( callback ! = null ) {
this . callback = callback ;
} else {
this . callback = TbMsgCallback . EMPTY ;
}
this . callback = Objects . requireNonNullElse ( callback , TbMsgCallback . EMPTY ) ;
}
public static ByteString toByteString ( TbMsg msg ) {
@ -284,11 +438,7 @@ public final class TbMsg implements Serializable {
public TbMsgCallback getCallback ( ) {
// May be null in case of deserialization;
if ( callback ! = null ) {
return callback ;
} else {
return TbMsgCallback . EMPTY ;
}
return Objects . requireNonNullElse ( callback , TbMsgCallback . EMPTY ) ;
}
public void pushToStack ( RuleChainId ruleChainId , RuleNodeId ruleNodeId ) {
@ -317,4 +467,18 @@ public final class TbMsg implements Serializable {
}
return ts ;
}
public boolean isTypeOf ( TbMsgType tbMsgType ) {
return tbMsgType ! = null & & tbMsgType . name ( ) . equals ( this . type ) ;
}
public boolean isTypeOneOf ( TbMsgType . . . types ) {
for ( TbMsgType type : types ) {
if ( isTypeOf ( type ) ) {
return true ;
}
}
return false ;
}
}
rusikv
3 years ago