public void run() {
            ByteBuffer buf=direct_buffers? ByteBuffer.allocateDirect(RoundTrip.PAYLOAD) : ByteBuffer.allocate(RoundTrip.PAYLOAD);
            for(;;) {
                try {
                    buf.position(0);
                    int num=client_channel.read(buf);
                    if(num == -1)
                        break;
                    if(num != RoundTrip.PAYLOAD)
                        throw new IllegalStateException("expected " + RoundTrip.PAYLOAD + " bytes, but got only " + num);
                    if(receiver != null) {
                        buf.flip();
                        int offset=buf.hasArray()? buf.arrayOffset() + buf.position() : buf.position(), len=buf.remaining();
                        if(!buf.isDirect())
                            receiver.receive(null, buf.array(), offset, len);
                        else { // by default use a copy; but of course implementers of Receiver can override this
                            byte[] tmp=new byte[len];
                            buf.get(tmp, 0, len);
                            receiver.receive(null, tmp, 0, len);
                        }
                    }
                }
                catch(Exception e) {
                    e.printStackTrace();
                }
            }
            Util.close(client_channel);
        }