path: root/src/org/gnunet/requests/RequestQueue.java

/*
 This file is part of GNUnet.
 (C) 2011, 2012 Christian Grothoff (and other contributing authors)

 GNUnet is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published
 by the Free Software Foundation; either version 3, or (at your
 option) any later version.

 GNUnet is distributed in the hope that it will be useful, but
 WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with GNUnet; see the file COPYING.  If not, write to the
 Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 Boston, MA 02111-1307, USA.
 */

package org.gnunet.requests;

import org.gnunet.util.*;

import java.util.LinkedList;

/**
 * Generic queues for Requests to be sent to the service.
 */
public class RequestQueue {

    /**
     * Requests to be transmitted to the service.
     */
    private final LinkedList<Request> requestsAwaitingTransmit = new LinkedList<Request>();

    /**
     * Persistent requests. Will be informed about reconnect / destroy events even
     * if already transmitted. Have to be canceled manually.
     */
    private final LinkedList<Request> persistentRequests = new LinkedList<Request>();

    /**
     * List of all requests from requestAwaitingTransmit and persistentRequest, containing no duplicates.
     */
    private final LinkedList<Request> allRequests = new LinkedList<Request>();

    /**
     * The designated receiver for all messages.
     */
    private MessageReceiver receiver;

    /**
     * The active transmit request handle, if any.
     */
    private Cancelable currentTransmit;

    /**
     * Current receive handler.
     */
    private Cancelable currentReceive;

    /**
     * True if we should not send further requests until queue is unclogged.
     */
    private boolean clogged = false;

    private boolean destroyed = false;
    private final Client client;

    public RequestQueue(Client client, MessageReceiver receiver) {
        this.client = client;
        this.receiver = receiver;
    }

    /**
     * Handle next request.
     */
    private void handleNextTransmit() {

        if (clogged) {
            return;
        }

        // return if we are already transmitting something
        if (currentTransmit != null) {
            return;
        }

        final Request request = requestsAwaitingTransmit.poll();
        if (request == null) {
            handleReceive();
            return;
        }

        AbsoluteTime deadline = request.getDeadline();
        if (deadline == null) {
            throw new AssertionError("getDeadline() must return a non-null AbsoluteTime");
        }

        currentTransmit = client.notifyTransmitReady(deadline.getRemaining(), true, 0, new MessageTransmitter() {
            @Override
            public void transmit(Connection.MessageSink sink) {
                currentTransmit = null;

                try {
                    request.transmit(sink);
                } finally {
                    handleReceive();
                    handleNextTransmit();
                }
            }

            @Override
            public void handleError() {
                throw new AssertionError("not implemented");
            }
        });
    }

    /**
     * Continue receiving if necessary.
     */
    private void handleReceive() {
        if (currentReceive != null || destroyed || !client.isConnected()) {
            return;
        }
        currentReceive = client.receive(RelativeTime.FOREVER, new MessageReceiver() {
            @Override
            public void process(GnunetMessage.Body msg) {
                currentReceive = null;

                try {
                    receiver.process(msg);
                } finally {
                    handleNextTransmit();
                    handleReceive();
                }
            }

            @Override
            public void handleError() {
                receiver.handleError();
            }
        });
    }

    /**
     * Add a request to the end of the queue.
     *
     * @param request request to be added
     * @return a handle to cancel the request
     */
    public Cancelable add(final Request request) {
        allRequests.add(request);
        requestsAwaitingTransmit.add(request);
        handleNextTransmit();

        return new Cancelable() {
            @Override
            public void cancel() {
                RequestQueue.this.requestsAwaitingTransmit.remove(request);
                RequestQueue.this.persistentRequests.remove(request);
                RequestQueue.this.allRequests.remove(request);
                request.onCancel(!requestsAwaitingTransmit.contains(request));
            }
        };
    }


    /**
     * Add a request so that it will get notified about reconnect/destroy events,
     * even if it already has been transmitted.
     */
    public Cancelable addPersistent(final Request request) {
        persistentRequests.add(request);
        return add(request);
    }


    /**
     * Add a request to the front of the queue, this request will be sent as
     * the next message (if not preempted by another sendNext).
     *
     * @param request request to be sent next
     * @return a handle to cancel the request
     */
    public Cancelable sendNext(final Request request) {
        requestsAwaitingTransmit.addFirst(request);
        handleNextTransmit();
        // todo: should this really return Cancelable? When do we want to cancel a request added by sendNext?
        return new Cancelable() {
            @Override
            public void cancel() {
                RequestQueue.this.requestsAwaitingTransmit.remove(request);
                RequestQueue.this.persistentRequests.remove(request);
                RequestQueue.this.allRequests.remove(request);
                request.onCancel(!requestsAwaitingTransmit.contains(request));
            }
        };
    }

    /**
     * Reconnect the client and notify all pending request of the reconnect.
     */
    public void reconnect() {
        client.reconnect();
        currentReceive = null;
        currentTransmit = null;

        final LinkedList<Request> remove = new LinkedList<Request>();


        for (Request r : allRequests) {
            boolean keep = r.onReconnect();
            if (!keep) {
                remove.add(r);
            } else {
                // retransmit an apparently persistent request.
                if (!requestsAwaitingTransmit.contains(r)) {
                    requestsAwaitingTransmit.add(r);
                }
            }
        }
        requestsAwaitingTransmit.removeAll(remove);
        persistentRequests.removeAll(remove);
        allRequests.removeAll(remove);

        // only transmit, receive should only be called after the first transmit
        handleNextTransmit();
    }

    /**
     * Notify all request of the shutdown. Does not actually destroy the connection.
     */
    public void shutdown() {
        final LinkedList<Request> remove = new LinkedList<Request>();

        for (Request r : allRequests) {
            boolean keep = r.onDestroy();
            if (!keep) {
                remove.add(r);
            } else {
                // retransmit an apparently persistent request.
                if (!requestsAwaitingTransmit.contains(r)) {
                    requestsAwaitingTransmit.add(r);
                }
            }
        }
        requestsAwaitingTransmit.removeAll(remove);
        persistentRequests.removeAll(remove);
        allRequests.removeAll(remove);

        handleNextTransmit();
        handleReceive();
    }

    /**
     * Cancel all requests and destroy the connection.
     */
    public void destroy() {
        destroyed = true;
        allRequests.clear();
        persistentRequests.clear();
        requestsAwaitingTransmit.clear();
        if (currentTransmit != null) {
            currentTransmit.cancel();
        }
        if (currentReceive != null) {
            currentReceive.cancel();
        }
    }

    /**
     * Allow no further requests to be transmitted until the queue is unclogged.
     */
    public void clog() {
        if (clogged) {
            throw new AssertionError("double clog");
        }
        clogged = true;
    }

    /**
     * Unclog the queue, must have been previously clogged.
     */
    public void unclog() {
        if (!clogged) {
            throw new AssertionError("unclogg before clog");
        }
        clogged = false;
        handleNextTransmit();
    }
}