private Collection < Message > peekMessages ( String shardName , int itemsToPeek ) throws MessageQueueException { try { ColumnList < MessageQueueEntry > result = keyspace . prepareQuery ( queueColumnFamily ) . setConsistencyLevel ( consistencyLevel ) . getKey ( shardName ) . withColumnRange ( new RangeBuilder ( ) . setLimit ( itemsToPeek ) . setStart ( entrySerializer . makeEndpoint ( ( byte ) MessageQueueEntryType . Message . ordinal ( ) , Equality . GREATER_THAN_EQUALS ) . toBytes ( ) ) . setEnd ( entrySerializer . makeEndpoint ( ( byte ) MessageQueueEntryType . Message . ordinal ( ) , Equality . LESS_THAN_EQUALS ) . toBytes ( ) ) . build ( ) ) . execute ( ) . getResult ( ) ; List < Message > messages = Lists . newArrayListWithCapacity ( result . size ( ) ) ; for ( Column < MessageQueueEntry > column : result ) { Message message = extractMessageFromColumn ( column ) ; if ( message != null ) messages . add ( message ) ; } return messages ; } catch ( ConnectionException e ) { throw new MessageQueueException ( "Error peeking for messages from shard " + shardName , e ) ; } }
Peek into messages contained in the shard . This call does not take trigger time into account and will return messages that are not yet due to be executed