path: root/src/main/java/org/gnunet/statistics/Statistics.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.statistics;

import org.gnunet.requests.MatchingRequestContainer;
import org.gnunet.requests.RequestIdentifier;
import org.gnunet.requests.SequentialRequestContainer;
import org.gnunet.requests.TimeoutHandler;
import org.gnunet.statistics.messages.GetResponseEndMessage;
import org.gnunet.statistics.messages.GetResponseMessage;
import org.gnunet.statistics.messages.SetMessage;
import org.gnunet.statistics.messages.WatchResponseMessage;
import org.gnunet.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * API for the GNUnet statistics service.
 * <p/>
 * Set, get and monitor statistics values, represented as unsigned 64bit integer.
 * Note that {@literal long}, java's largest primitive type, can only store signed 64bit integers.
 * With absolute operation, its negative values are interpreted as large numbers by the statistics api.
 */
public class  Statistics {
    private static final Logger logger = LoggerFactory
            .getLogger(Statistics.class);

    /**
     * Client connecting us to the statistics service.
     */
    private Client client;

    /**
     * All request to the service for getting a value.
     */
    private final SequentialRequestContainer<GetRequest> getRequests;

    /**
     * All requests to the service for watching a value.
     */
    private final MatchingRequestContainer<Long,WatchRequest> watchRequests;

    /**
     * Do we wait for the final 'TestMessage' from the service and
     * do not accept any new requests?
     */
    private boolean destroyRequested;

    /**
     * Next unused ID to identity watch requests/responses.
     */
    private long nextWatchId = 0;

    /**
     * Timeout when waiting for the TestMessage after destruction of
     * this statistics handle has been requested.
     */
    private Scheduler.TaskIdentifier destroyTimeout;

    /**
     * Messages from the statistics service are dispatched to an instance of this class.
     */
    public class StatisticsMessageReceiver extends RunaboutMessageReceiver {
        public void visit(GetResponseMessage m) {
            RequestIdentifier<GetRequest> r = getRequests.getRequestIdentifier();
            if (r != null)
	            r.getRequest().receiver.onReceive(m.subsystemName, m.statisticName, m.value);
        }

        public void visit(@SuppressWarnings("UnusedParameters") GetResponseEndMessage m) {
            RequestIdentifier<GetRequest> r = getRequests.getRequestIdentifier();
            if (r != null) {
                r.retire();
                r.getRequest().receiver.onDone();
            }
        }

        public void visit(@SuppressWarnings("UnusedParameters") TestMessage m) {
            // The TestMessage indicates that the statistics service received all our
            // messages, we can disconnect.
            if (null != destroyTimeout) {
                destroyTimeout.cancel();
                destroyTimeout = null;
            } else {
                logger.error("protocol violation: destroy timeout is 'null' but got test message");
            }
            client.disconnect();
        }

        public void visit(WatchResponseMessage wrm) {
            RequestIdentifier<WatchRequest> ri = watchRequests.getRequestIdentifier((long) wrm.wid);
            WatchRequest r = ri.getRequest();
            if (r != null) {
                r.watcher.onReceive(r.subsystem, r.name, wrm.value);
            }
        }

        @Override
        public void handleError() {
            if (null == client)
                throw new AssertionError();
            if (!destroyRequested) {
                client.reconnect();
                getRequests.restart();
                watchRequests.restart();
            }
            // if everything is shutting down, maybe the statistics service
            // was shut down, and can't respond with the TestMessage anymore.
            if (null != destroyTimeout) {
                destroyTimeout.cancel();
                destroyTimeout = null;
            }
        }
    }

    /**
     * Create a connection to the statistics service.
     *
     * @param cfg configuration to use
     */
    public Statistics(Configuration cfg) {
        client = new Client("statistics", cfg);
        client.installReceiver(new StatisticsMessageReceiver());
        getRequests = new SequentialRequestContainer<GetRequest>(client);
        watchRequests = new MatchingRequestContainer<Long, WatchRequest>(client);
    }

    /**
     * Retrieve a statistics value of a subsystem.
     *
     * @param timeout      time after we give up and call receiver.onTimeout
     * @param subsystem    the subsystem of interest
     * @param name         name of the statistics value belongs to
     * @param receiver     callback
     * @return handle to onCancel the getRequestIdentifier
     */
    public Cancelable get(RelativeTime timeout, final String subsystem, final String name,
                          final StatisticsReceiver receiver) {
        if (destroyRequested || client == null)
            throw new AssertionError("already destroyed");
        RequestIdentifier<GetRequest> identifier = getRequests.addRequest(new GetRequest(subsystem, name, receiver));
        identifier.setTimeout(timeout, new TimeoutHandler() {
            @Override
            public void onTimeout() {
                receiver.onTimeout();
            }
        });
        return identifier;
    }

    /**
     * Retrieve all statistics value of a subsystem.
     *
     * @param timeout      time after we give up and call receiver.onTimeout
     * @param subsystem    the subsystem of interest
     * @param receiver     callback
     * @return handle to onCancel the getRequestIdentifier
     */
    public Cancelable get(RelativeTime timeout, final String subsystem,
                          final StatisticsReceiver receiver) {
        return get(timeout, subsystem, "", receiver);
    }

    /**
     * Sets a statistics value asynchronously.
     *
     * @param name    name of the entry
     * @param value   desired value
     * @param persist keep value even if the statistics service restarts
     */
    public void set(final String subsystem, final String name, final long value, boolean persist) {
        if (destroyRequested || client == null)
            throw new AssertionError("already destroyed");
        SetMessage m = new SetMessage();
        m.statisticName = name;
        m.subsystemName = subsystem;
        m.value = value;
        if (persist)
            m.flags |= SetMessage.SETFLAG_PERSIST;
        client.send(m);
    }

    /**
     * Changes a statistics value asynchronously.
     *
     * @param name    name of the entry
     * @param delta   relative difference to the old value
     * @param persist keep value even if the statistics service restarts
     */
    public void update(final String subsystem, final String name, final long delta, boolean persist) {
        if (destroyRequested || null == client)
            throw new AssertionError("already destroyed");
        SetMessage m = new SetMessage();
        m.statisticName = name;
        m.subsystemName = subsystem;
        m.value = delta;
        if (persist)
            m.flags |= SetMessage.SETFLAG_PERSIST;
        m.flags |= SetMessage.SETFLAG_RELATIVE;
        client.send(m);
    }

    /**
     * Receive updates about changing statistics values.
     *
     * @param subsystem the subsystem to watch
     * @param name the value to watch
     * @param watcher the object that receives the updates
     * @return a handle to onCancel the getRequestIdentifier
     */
    public Cancelable watch(final String subsystem, final String name, StatisticsWatcher watcher) {
        if (destroyRequested || null == client)
            throw new AssertionError("already destroyed");
        WatchRequest r = new WatchRequest(subsystem, name, watcher);
        return watchRequests.addRequest(nextWatchId++, r);
    }


    /**
     * Destroy handle to the statistics service. Always finishes writing pending values.
     */
    public void destroy() {
      destroy(true);
    }

    /**
     * Destroy handle to the statistics service.
     *
     * @param syncFirst If true, wait until the statistics service has received all our updates.
     *                  If false, pending updates may be lost.
     */
    public void destroy(boolean syncFirst) {
        if (destroyRequested)
            throw new AssertionError("already destroyed");
        destroyRequested = true;
        logger.debug("destroying statistics");
        if (!syncFirst || !client.isConnected()) {
            client.disconnect();
            client = null;
            return;
        }
        client.send(new TestMessage());
        // wait until the service responds or a timeout occurs
        destroyTimeout = Scheduler.addDelayed(RelativeTime.fromSeconds(5), new Scheduler.Task() {
            @Override
            public void run(Scheduler.RunContext ctx) {
                if (null == client)
                    return;
                client.disconnect();
                client = null;
            }
        });
    }
}