.
Matching
Abstract Entity Matcher
Definition
abstract class EntityMatcherComp[M1: TypeTag, M2: TypeTag, OUT: TypeTag] extends ComponentContract
class EntityMatcher(implicit M1CTag: ClassTag[M1], M2CTag: ClassTag[M2]) extends Cell[EntityMatcherConfig]
Description
This is a parametrised component matching two streams of data. M1 and M2 represent the two types of message that will be received and matched one against the other.
The config contains two lists of key fields to extract from the two messages, m1Keys and m2Keys. Both sets of keys must be of the same size. The contained keys will be compared from one list to the other by position and so the key fields in the lists must be comparable from one list to the other by position in the list. The list’s keys are used to perform internal structure look ups of the values to compare.
The order and number of keys might impact performance as the matcher will keep lists of messages which have the same key values and it will apply the matchRules to all those messages in the list sequentially when looking for a data match.
The keys, therefore constitute an exact match and matchRules might contain extra custom matching rules, not necessarily full equality but satisfying a boolean expression for an adequate match.
The matcher needs to know when to consider messages unmatchable and that is given by the last parameter: a timeout period representing the elapsed time since a message was received to consider that it must be removed from consideration by being purged (purgeOlderThan). This time is taken as a minimum so messages discarded are as old as this time or older
The behaviour of the matcher, when a multiple match has been found, is to pick the first match in the set of matching messages.
Type parameters:
M1: Message type one
M2: Message type two
Contract
def contract: Contract =
contract(Receives[M1], Receives[M2], Sends[M1], Sends[M2], Sends[OUT])Configuration
def m1Keys: M1 => List[Any] // Key mapping function returning a list of key values
def m2Keys: M2 => List[Any] // Key mapping function returning a list of key values
def matchCondition: (M1, M2) => Boolean // The match function to apply to values matched by key
def outputFunction: (M1, M2) => OUT // The output value generation function for matched values
def m1Behavior: Behavior = Behavior() // Sets the match behaviour for M1 messages
def m2Behavior: Behavior = Behavior() // Sets the match behaviour for M2 messages
def flushOnEndData: Boolean = false // Flag to determine if matcher is flushed on end op.Usage
abstract class TupleEntityMatcherComp[M1: TypeTag, M2: TypeTag] extends EntityMatcherComp[M1, M2, (M1, M2)]
object TradeMatcherComp extends TupleEntityMatcherComp[ClientTrade, BrokerTrade]Abstract Tuple Entity Matcher
Definition
abstract class TupleEntityMatcherComp[M1: TypeTag, M2: TypeTag] extends EntityMatcherComp[M1, M2, (M1, M2)]Description
A component to present matched entities as tuples of the matching pairs.
Contract
def contract: Contract =
contract(Receives[M1], Receives[M2], Sends[(M1, M2)], Sends[OUT])Configuration
override def outputFunction: (M1, M2) => (M1, M2) = (m1, m2) => (m1, m2)Usage
object TradeMatcherComp extends TupleEntityMatcherComp[ClientTrade, BrokerTrade]Abstract Left Joiner
Definition
abstract class LeftJoinerComp[L: TypeTag, R: TypeTag, M: TypeTag] extends ComponentContract
class LeftJoiner(implicit LCTag: ClassTag[L], RCTag: ClassTag[R]) extends Cell[LeftJoinerConfig]Description
This is a parametrised component for joining two streams of data. L and R represent the two types of message that will be received and matched one against the other (Left and Right).
The config contains a list of keys for each stream (leftKeys and rightKeys) so that values can be extracted from the two messages to compare as keys. Both lists of keys must be the same length and have the same positional order for each field from each entity type contributing to the key based (complete) match.
The order and number of keys might impact performance as the joiner will retain messages which have the same key values and it will apply the join condition to all those messages in the list sequentially when looking up a match.
Think about the keys as the ON clause matching ids on a DB join, and the join condition like the additional filtering condition which can be anything satisfying a boolean expression.
If the component is configured as leftNative (setting leftNative to true which is the default) then it expects to be preload with L messages and then receive the stream of R messages which will be matched against the pre-loaded L messages. Otherwise, if leftNative is set to false, the component will expect all R messages first and then to match R appropriately.
In leftNative mode L messages are held within the component data structures and R messages arriving to the component are matched or discarded. When a match happens L messages may be discarded or kept depending on cardinality configuration.
The component expects to participate in two phases led by 2 operations. The first phase will load one side of the join, the second will match the other side against loaded items of the other side. The end of the Join is reached upon receiving the second End of Data Confirmation signal (for the second phase), at this point all unmatched L messages remaining will be sent out as unmatched.
The output type of the component when a join match is produced can also be configured by binding type parameter M and providing an output function like (L, R) => M. Similarly, left messages remaining unmatched and right messages being discarded will be sent out. User can choose to route them to “Nowhere” if not required.
The cardinality of the left – right relationship can be specified by setting the configuration cardinality to
OneOne, OneMany, ManyOne or ManyMany. This will affect performance as items of cardinality One will be discarded immediately after a match is produced. However if the cardinality setting does not match the real cardinality of the data then the component might yield unexpected results.
When the component discard messages they are removed from the internal structures and not sent away. When unmatched messages are flushed they are removed from internal structures and sent out individually.
Type parameters:
L (Left message type)
R (Right message type)
M Out type on a join match
Contract
def contract: Contract = contract(Receives[L], Receives[R], Sends[L], Sends[R], Sends[M])
Configuration
def leftKeys: L => List[Any] // Left value extraction function
def rightKeys: R => List[Any] // Right value extraction function
def joinCondition: (L, R) => Boolean = (_, _) => true // Join condition predicate
def matchOutFn: (L, R) => M // Output value function
def leftNative: Boolean = true // Join type left or right
def cardinality: Cardinality = ManyMany // Cardinality Usage
object TuplesLeftJoinerComp extends LeftJoinerComp[(Int, String), (Int, String, Double), (Int, String, String, Double)]Abstract Entity Comparator
Definition
abstract class EntityComparatorComp[T1:TypeTag, T2:TypeTag, DIFF:TypeTag] extends ComponentContract
abstract class EntityComparator[C <: Config] extends Cell[C]Description
General purpose Entity Comparator base component.
This class of components receive pairs of entities to be compared and send them out untouched if they satisfy a comparison function provided in the config. If differences exist the component will send out a DIFF message which will represent the differences between the types
Type parameters:
T1 One of the types to be compared
T2 The other type to be compared
DIFF A type representing the differences.
Contract
override def contract = contract(Receives[(T1, T2)], Sends[(T1, T2)], Sends[DIFF])Configuration
def compareCondition: (T1, T2) => Option[DIFF] = (_, _) => None // Comparison output function
def sendOutDifferent: Boolean = false // Flag to enable diff outputUsage
object RecordABComparatorComp extends EntityComparatorComp[RecordA, RecordB, RecordDifferences]{
trait Config extends super.Config {}
class RecordABComparator extends EntityComparator
}Record Comparator
Definition
object RecordComparatorComp extends EntityComparatorComp[Record, Record, RecordDifferences]
class RecordComparator extends EntityComparatorDescription
A Record comparison component derived from EntityComparator.
The record comparison yields a RecordDifference type containing a reference to the Record instances compared and a collection of FieldDifference values to describe the differences by value and index.
Contract
Inherited from EntityComparator
Configuration
Overrides to parent config:
override def compareCondition: (Record, Record) => Option[RecordDifferences] // The comparison function
override def sendOutDifferent: Boolean = false // Output control flag Usage
object RecordComparatorComp extends EntityComparatorComp[Record, Record, RecordDifferences]{
trait Config extends super.Config {
override def compareCondition: (Record, Record) => Option[RecordDifferences]
override def sendOutDifferent: Boolean = false
}
class RecordComparator extends EntityComparator
}