@ Override public ReceiveQueue < T > receiveQueue ( ) { final ReceiveQueue < String > receiveQueue = queue . receiveQueue ( ) ; return new ReceiveQueue < T > ( ) { @ Override public T pollWait ( ) { final String item = receiveQueue . pollWait ( ) ; return getParsedItem ( item ) ; } private T getParsedItem ( String item ) { if ( item != null ) { return fromJsonFunction . apply ( item ) ; } else { return null ; } } @ Override public T poll ( ) { final String item = receiveQueue . pollWait ( ) ; return getParsedItem ( item ) ; } @ Override public T take ( ) { final String item = receiveQueue . take ( ) ; return getParsedItem ( item ) ; } @ Override public Iterable < T > readBatch ( int max ) { final Iterable < String > iterable = receiveQueue . readBatch ( max ) ; return getParsedItems ( iterable ) ; } private Iterable < T > getParsedItems ( Iterable < String > iterable ) { int size = 16 ; if ( iterable instanceof List ) { size = ( ( List ) iterable ) . size ( ) ; } final List < T > items = new ArrayList <> ( size ) ; for ( String item : iterable ) { items . add ( getParsedItem ( item ) ) ; } return items ; } @ Override public Iterable < T > readBatch ( ) { final Iterable < String > iterable = receiveQueue . readBatch ( ) ; return getParsedItems ( iterable ) ; } @ Override public void stop ( ) { receiveQueue . stop ( ) ; } } ; }
Create a wrapper ReceiveQueue that does decoding on the fly .