@Override
    protected void process_activate_write (long msgs_read_)
    {
        //  Remember the peers's message sequence number.
        peers_msgs_read = msgs_read_;

        if (!out_active && state == State.active) {
            out_active = true;
            sink.write_activated (this);
        }
    }