protected MessageTuple getMessage ( int timeoutMs ) throws InterruptedException , TimeoutException { MessageTuple tuple = messageQueue . poll ( timeoutMs , TimeUnit . MILLISECONDS ) ; if ( tuple == null ) { throw new TimeoutException ( "Timeout while waiting for the message." ) ; } if ( tuple . getMessage ( ) . getType ( ) == MessageType . SHUT_DOWN ) { // If it's SHUT_DOWN message. throw new LeftCluster ( "Left cluster" ) ; } return tuple ; }
Takes the filtered message from message queue . This method will be blocked until the message becomes available or timeout is detected .