Pre instantiation

Message Processor Pre-Instantiation

In some cases your use-case may call for the pre-instantiation (or "explicit instantiation") of all of the message processors in a cluster in order for them to participate in an output cycle. Dempsy provides you with a hook where you can provide the entire key space (the set of all potential message keys for a class of messages) to the framework and as part of the initialization of the container, Dempsy will pre-instantiate all of the corresponding message processors.

This is accomplished by supplying a custom KeySource to the ClusterDefinition. A KeySource is an interface with a single method:

     public Iterable<T> getAllPossibleKeys();

Note: In version 0.6 "KeySource" is called "KeyStore." The name is changed in 0.7 for clarity.

Let's suppose you want to keep track of the pieces on a checker board where the a message is emitted whenever a piece lands on a square, or leaves a square. That message could look something like this:

   public class CheckerBoardSquareState
   {
      public static final int checkerBoardWidth = 8;
      public static enum State { on, off };
      
      private State state;
      private Integer positionOnBoard = -1;
      
      public CheckerBoardSquareState(int row, int col, State state)
      {
         this.positionOnBoard = (row * checkerBoardWidth) + col;
         this.state = state;
      }
      
      @MessageKey
      public Integer getPositionOnBoard() { return positionOnBoard; }
      
      public State getState() { return state; }
      
      public int getRow() { return checkerBoardWidth / positionOnBoard; }
      public int getColumn() { return checkerBoardWidth % positionOnBoard; }
   }

We're assuming that the CheckerBoardSquare message is emitted from an Adaptor for a particular row and column on a checkerboard and contains the State which is an indication of whether or not that place on the board has just had a piece placed there (on) or removed from there (off).

Now, suppose I want a message processor that keeps track of the state of the squares, and periodically emits a message for a later processing stage. That message processor could look like:

   @MessageProcessor
   public class CheckerBoardMonitorMp
   {
      int numStateChanges = 0;
      CheckerBoardSquareState.State currentState = CheckerBoardSquareState.State.off;
      int positionOnBoard = -1;
      
      @Activation
      public void setPositionOnBoard(Integer positionOnBoard)
      {
         this.positionOnBoard = positionOnBoard;
      }
      
      @MessageHandler
      public void stateChanged(CheckerBoardSquareState newState)
      {
         numStateChanges++;
         currentState = newState.getState();
      }
      
      @Output
      public CheckerBoardPositionStatus getCurrentPositionStatus()
      {
          // Here we emit a new message called CheckerBoardPositionStatus 
          // in order to do further processing downstream. CheckerBoardPositionStatus
          // contains the current state, the position, and all the current count.
          ...
      }
   }

It may be the case that the downstream processing should receive @Output messages from the CheckerBoardMonitorMp even if there hasn't been any messages received for that particular row and column of the checker board yet.

In order to facilitate this we need to "pre-instantiate" all of the CheckerBoardMonitorMp message processors that cover the checker board. This can be done by supplying a KeySource that can tell Dempsy all of the message keys who's corresponding message processors should be pre-instantiated in the container.

For our example this could look something like:

   public class CheckerBoardKeySpace implements KeySource<Integer>
   {
      private static Integer[] keyspaceArray = new Integer[64];
      private static List<Integer> keyspace;
      
      static {
         for (int i = 0; i < 64; i++) keyspaceArray[i] = i;
         keyspace = Collections.unmodifiableList(Arrays.asList(keyspaceArray));
      }
      
      @Override
      public Iterable<Integer> getAllPossibleKeys()
      {
         return keyspace;
      }
   }

In order to supply it to Dempsy you use the ClusterDefinition. Programatically you would create the ClusterDefinition using something like the following:

      ...
      new ClusterDefinition("checker-board-monitor", new CheckerBoardMonitorMp()).setKeySource(new CheckerBoardKeySpace())
      ...

In Spring you would do something like:

   <bean class="com.nokia.dempsy.config.ClusterDefinition">
     <constructor-arg value="checker-board-monitor"/>
     <property name="messageProcessorPrototype">
       <bean class="com.nokia.dempsy.example.checkerboard.CheckerBoardMonitorMp"/>
     </property>
     <property name="keySource">
       <bean class="com.nokia.dempsy.example.checkerboard.CheckerBoardKeySpace"/>
     </property>
   </bean>

Now, when Dempsy starts it will work through the key space and instantiate each corresponding message processor.

Important things to keep in mind:

• Dempsy does NOT hold up message processing in order to instantiate the entire key space of message processors. That is, the message processor instantiation will take place concurrently with the handling of messages. Therefore, it is possible that an message processor is instantiated due to a received message prior to the explicit instantiation due to the KeySource.
• When running distributed, Dempsy is smart enough to ignore keys in the key space that are not meant for the node where the pre-instantiation is executing. That is, even though the entire key space will be iterated in each container in a cluster, not every corresponding message processor will be instantiated if the current container is not responsible for handling that message key.

Another way to look at this is to consider what happens when the above example is run distributed over 8 nodes. The container in each node will iterate over all 64 keys, however, each node will only have 8 message processors instantiated, for a total of 64 across the cluster.

Next section: Eviction