@ SuppressWarnings ( "unchecked" ) private void consume ( LogRecord record ) { // Decode the raw log operation from the record. LogOperation operation = decodeInternal ( record . value ( ) ) ; // If this operation is not destined for this primitive, ignore it. // TODO: If multiple primitives of different types are created and destroyed on the same distributed log, // we need to be able to differentiate between different instances of a service by the service ID. if ( ! operation . primitive ( ) . equals ( name ( ) ) ) { return ; } // Create a session from the log record. Session session = getOrCreateSession ( operation . sessionId ( ) ) ; // Update the local context for the service. currentIndex = record . index ( ) ; currentSession = session ; currentOperation = operation . operationId ( ) . type ( ) ; currentTimestamp = record . timestamp ( ) ; // Apply the operation to the service. byte [ ] output = service . apply ( new DefaultCommit <> ( currentIndex , operation . operationId ( ) , operation . operation ( ) , currentSession , currentTimestamp ) ) ; // If the operation session matches the local session, complete the write future. if ( operation . sessionId ( ) . equals ( this . session . sessionId ( ) ) ) { CompletableFuture future = writeFutures . remove ( operation . operationIndex ( ) ) ; if ( future != null ) { future . complete ( decode ( output ) ) ; } } // Iterate through pending reads and complete any reads at indexes less than or equal to the applied index. PendingRead pendingRead = pendingReads . peek ( ) ; while ( pendingRead != null && pendingRead . index <= record . index ( ) ) { session = getOrCreateSession ( this . session . sessionId ( ) ) ; currentSession = session ; currentOperation = OperationType . QUERY ; try { output = service . apply ( new DefaultCommit <> ( currentIndex , pendingRead . operationId , pendingRead . bytes , session , currentTimestamp ) ) ; pendingRead . future . complete ( output ) ; } catch ( Exception e ) { pendingRead . future . completeExceptionally ( new PrimitiveException . ServiceException ( ) ) ; } pendingReads . remove ( ) ; pendingRead = pendingReads . peek ( ) ; } }
Consumes a record from the log .