@Override
    public void handleReply(Reply reply) {
        if (failure.get() != null) {
            return;
        }
        if (containsFatalErrors(reply.getErrors())) {
            failure.compareAndSet(null, new IOException(formatErrors(reply)));
            return;
        }
        long now = System.currentTimeMillis();
        long latency = now - (long) reply.getContext();
        numReplies.incrementAndGet();
        accumulateReplies(now, latency);
    }
@Test
    public void requireThatJsonFeederWorks() throws Throwable {
        assertFeed("[" +
                        "  { \"put\": \"id:simple:simple::0\", \"fields\": { \"my_str\":\"foo\"}}," +
                        "  { \"update\": \"id:simple:simple::1\", \"fields\": { \"my_str\": { \"assign\":\"bar\"}}}," +
                        "  { \"remove\": \"id:simple:simple::2\"}" +
                        "]",
                new MessageHandler() {

                    @Override
                    public void handleMessage(Message msg) {
                        Reply reply = ((DocumentMessage)msg).createReply();
                        reply.swapState(msg);
                        reply.popHandler().handleReply(reply);
                    }
                },
                "",
                "(.+\n)+" +
                        "\\s*\\d+,\\s*3,.+\n");
    }