qbit/core/src/main/java/io/advantageous/qbit/queue/JsonQueue.java
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.