private SendQueue<T> createJsonSendQueue(final SendQueue<String> sendQueue) {

        return new SendQueue<T>() {
            @Override
            public boolean send(T item) {

                sendQueue.send(toJsonFunction.apply(item));
                return false;
            }

            @Override
            public void sendAndFlush(T item) {

                sendQueue.sendAndFlush(toJsonFunction.apply(item));
            }

            @Override
            public void sendMany(T... items) {

                for (T item : items) {
                    sendQueue.send(toJsonFunction.apply(item));
                }
            }

            @Override
            public void sendBatch(Collection<T> items) {

                for (T item : items) {
                    sendQueue.send(toJsonFunction.apply(item));
                }
            }

            @Override
            public void sendBatch(Iterable<T> items) {

                for (T item : items) {
                    sendQueue.send(toJsonFunction.apply(item));
                }
            }

            @Override
            public boolean shouldBatch() {
                return sendQueue.shouldBatch();
            }

            @Override
            public void flushSends() {

                sendQueue.flushSends();
            }

            @Override
            public int size() {
                return sendQueue.size();
            }

            @Override
            public String name() {
                return sendQueue.name();
            }

            @Override
            public void stop() {
                sendQueue.stop();
            }
        };
    }
private SendQueue<T> createJsonSendQueue(final SendQueue<String> sendQueue) {

        return new SendQueue<T>() {
            @Override
            public boolean send(T item) {

                sendQueue.send(toJsonFunction.apply(item));
                return false;
            }

            @Override
            public void sendAndFlush(T item) {

                sendQueue.sendAndFlush(toJsonFunction.apply(item));
            }

            @Override
            public void sendMany(T... items) {

                for (T item : items) {
                    sendQueue.send(toJsonFunction.apply(item));
                }
            }

            @Override
            public void sendBatch(Collection<T> items) {

                for (T item : items) {
                    sendQueue.send(toJsonFunction.apply(item));
                }
            }

            @Override
            public void sendBatch(Iterable<T> items) {

                for (T item : items) {
                    sendQueue.send(toJsonFunction.apply(item));
                }
            }

            @Override
            public boolean shouldBatch() {
                return sendQueue.shouldBatch();
            }

            @Override
            public void flushSends() {

                sendQueue.flushSends();
            }

            @Override
            public int size() {
                return sendQueue.size();
            }

            @Override
            public String name() {
                return sendQueue.name();
            }

            @Override
            public void stop() {
                sendQueue.stop();
            }
        };
    }
private SendQueue < T > createJsonSendQueue ( final SendQueue < String > sendQueue ) { return new SendQueue < T > ( ) { @ Override public boolean send ( T item ) { sendQueue . send ( toJsonFunction . apply ( item ) ) ; return false ; } @ Override public void sendAndFlush ( T item ) { sendQueue . sendAndFlush ( toJsonFunction . apply ( item ) ) ; } @ Override public void sendMany ( T ... items ) { for ( T item : items ) { sendQueue . send ( toJsonFunction . apply ( item ) ) ; } } @ Override public void sendBatch ( Collection < T > items ) { for ( T item : items ) { sendQueue . send ( toJsonFunction . apply ( item ) ) ; } } @ Override public void sendBatch ( Iterable < T > items ) { for ( T item : items ) { sendQueue . send ( toJsonFunction . apply ( item ) ) ; } } @ Override public boolean shouldBatch ( ) { return sendQueue . shouldBatch ( ) ; } @ Override public void flushSends ( ) { sendQueue . flushSends ( ) ; } @ Override public int size ( ) { return sendQueue . size ( ) ; } @ Override public String name ( ) { return sendQueue . name ( ) ; } @ Override public void stop ( ) { sendQueue . stop ( ) ; } } ; }
Helper method to create SendQueue wrappers that do JSON encoding.

@param sendQueue sendQueue
@return sendQueue decorated with JSON encoding.
Helper method to create SendQueue wrappers that do JSON encoding .