private Observable < PartitionAndSeqno > getSeqnos ( ) { return conductor . getSeqnos ( ) . flatMap ( new Func1 < ByteBuf , Observable < PartitionAndSeqno > > ( ) { @ Override public Observable < PartitionAndSeqno > call ( ByteBuf buf ) { int numPairs = buf . readableBytes ( ) / 10 ; // 2 byte short + 8 byte long List < PartitionAndSeqno > pairs = new ArrayList <> ( numPairs ) ; for ( int i = 0 ; i < numPairs ; i ++ ) { pairs . add ( new PartitionAndSeqno ( buf . getShort ( 10 * i ) , buf . getLong ( 10 * i + 2 ) ) ) ; } buf . release ( ) ; return Observable . from ( pairs ) ; } } ) ; }
Get the current sequence numbers from all partitions .