Overview
The Ordered Service is required by the Ordered Routing Queue Strategy. A function defined in APL code will extract values from any routed UDR or bytearray to decide which partition queue to add it to.
Ordered Routing allows the system to utilize multicore cpus, but still have deterministic order.
Configuration of the Ordered Service
If there are session updates from a collection point that are encoded to a user-defined UDR type Ordered.Subscriber.Update with the field 'id' containing a session id, the following example shows how the values are read and added to the ordered.SessionIdentifier that the queue strategy will use to hash and select a partition based on.
import ultra.Ordered.Subscriber;
void route(ordered.SessionIdentifier si, any input) {
if(instanceOf(input, Update)) {
Update u = (Update)input;
ordered.addInteger(si, u.id);
}
}
Multiple values can be added to the SessionIdentifier, it will calculate the hash based on them all after the route function returns. If there is more than one type routed into the workflow, there must be a longer if-else chain, with a clause for each type.
Function | Description |
|---|---|
ordered.addInteger(ordered.SessionIdentifier si, int i) | Add the value of integer i to the |
ordered.addString(ordered.SessionIdentifier si, string s) | Add the value of string s to the |
ordered.addByte(ordered.SessionIdentifier si, byte b) | Add the value of byte b to the |
ordered.addBytes(ordered.SessionIdentifier si, bytearray ba) | Add the value of bytearray ba to the |
ordered.addByteRange(ordered.SessionIdentifier si, bytearray bs, int offset, int len) | Add the values of bytearray ba from offset and the next len bytes to the |
ordered.setPartition(ordered.SessionIdentifier si, int partition) | Alternatively, if more control over partition selection is wanted one can set the partition id explicitly, this will override any values added to the |
0 Comments