Dépôt officiel du code source de l'ERP OpenConcerto

/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 * 
 * Copyright 2011 OpenConcerto, by ILM Informatique. All rights reserved.
 * 
 * The contents of this file are subject to the terms of the GNU General Public License Version 3
 * only ("GPL"). You may not use this file except in compliance with the License. You can obtain a
 * copy of the License at http://www.gnu.org/licenses/gpl-3.0.html See the License for the specific
 * language governing permissions and limitations under the License.
 * 
 * When distributing the software, include this License Header Notice in each file.
 */
 
 package org.openconcerto.utils;

import org.openconcerto.utils.cc.IClosure;

import java.util.Deque;
import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;

import net.jcip.annotations.GuardedBy;
import net.jcip.annotations.ThreadSafe;

/**
 * Holds items and give them one at a time to {@link #process(Object)}. At any time this process can
 * be put on hold by setting it to sleep. All actions in a thread modifying ({@link #put(Object)},
 * {@link #itemsDo(IClosure)}) the queue happens-before the processing of this modification.
 * 
 * @author Sylvain
 * @param <T> type of item
 */
@ThreadSafe
public abstract class DropperQueue<T> extends Thread {

    @GuardedBy("itemsLock")
    private final Deque<T> items;
    private final Lock itemsLock;
    private final Condition notEmpty;
    @GuardedBy("this")
    private boolean stop;
    @GuardedBy("this")
    private boolean sleeping;
    @GuardedBy("this")
    private boolean executing;

    /**
     * Construct a new instance.
     * 
     * @param name name of this thread.
     */
    public DropperQueue(String name) {
        super(name);
        this.items = new LinkedList<T>();
        this.stop = false;
        this.sleeping = false;
        this.executing = false;
        this.itemsLock = new ReentrantLock();
        this.notEmpty = this.itemsLock.newCondition();
    }

    // *** boolean

    /**
     * Whether this queue should stop temporarily.
     * 
     * @param b <code>true</code> to put this to sleep, <code>false</code> to wake it.
     * @return <code>true</code> if sleeping has changed.
     */
    public synchronized final boolean setSleeping(boolean b) {
        if (this.sleeping != b) {
            this.sleeping = b;
            this.signalChange();
            return true;
        } else
            return false;
    }

    public synchronized boolean isSleeping() {
        return this.sleeping;
    }

    private synchronized void signalChange() {
        this.signalChange(false);
    }

    private synchronized void signalChange(boolean signalClosure) {
        // interrompt la thread, si elle est dans le await() ou le take()
        // elle recheckera les booleens, ATTN elle peut-être en attente du lock juste après take(),
        // c'est pourquoi on efface le flag avant process()
        // en général pas besoin d'interrompre la closure, puisque tant qu'on l'exécute
        // on ne peut, ni on n'a besoin de prendre depuis la queue
        if (signalClosure || !this.isExecuting())
            this.interrupt();
    }

    private synchronized void setExecuting(boolean b) {
        this.executing = b;
    }

    private synchronized boolean isExecuting() {
        return this.executing;
    }

    private void await() throws InterruptedException {
        synchronized (this) {
            if (this.sleeping) {
                this.wait();
            }
        }
    }

    /**
     * Signal that this thread must stop indefinitely. This method interrupts
     * {@link #process(Object)}.
     * 
     * @see #die(boolean)
     */
    public final void die() {
        this.die(true);
    }

    /**
     * Signal that this thread must stop indefinitely. Once this method returns, it is guaranteed
     * that no new item will be processed, and that this thread will {@link #isDead() die}. But if
     * this thread is currently {@link #process(Object) processing} an item, then the method will
     * finish normally if :
     * <ul>
     * <li><code>mayInterruptIfRunning</code> is <code>false</code></li>
     * <li><code>mayInterruptIfRunning</code> is <code>true</code> but the {@link #interrupt()}
     * isn't checked by the implementing subclass</li>
     * </ul>
     * 
     * @param mayInterruptIfRunning <code>true</code> to interrupt while in {@link #process(Object)}
     * @see #isDying()
     */
    public synchronized final void die(boolean mayInterruptIfRunning) {
        this.stop = true;
        this.signalChange(mayInterruptIfRunning);
    }

    /**
     * Whether this queue is dying.
     * 
     * @return <code>true</code> if {@link #die(boolean)} has been called and
     *         {@link #process(Object)} is still executing.
     * @see #isDead()
     */
    public synchronized final boolean isDying() {
        return this.dieCalled() && this.isExecuting();
    }

    /**
     * Whether this queue is active.
     * 
     * @return <code>true</code> if {@link #process(Object)} isn't executed and won't ever be again.
     * @see #isDying()
     */
    public final boolean isDead() {
        // either we're dead because die() has been called, or because process() threw an Error
        return this.getState().equals(State.TERMINATED);
    }

    public synchronized final boolean dieCalled() {
        return this.stop;
    }

    // *** Run

    @Override
    public void run() {
        while (!this.dieCalled()) {
            try {
                this.await();
                final T item;
                // lockInterruptibly() to avoid taking another item after being put to sleep
                this.itemsLock.lockInterruptibly();
                try {
                    while (this.items.isEmpty())
                        this.notEmpty.await();
                    item = this.items.removeFirst();
                } finally {
                    this.itemsLock.unlock();
                }
                this.setExecuting(true);
                // we should not carry the interrupted status in process()
                // we only use it to stop waiting and check variables again, but if we're here we
                // have removed an item and must process it
                Thread.interrupted();
                process(item);
            } catch (InterruptedException e) {
                // rien a faire, on recommence la boucle
            } catch (RuntimeException e) {
                e.printStackTrace();
                // une exn s'est produite, on considère qu'on peut passer à la suite
            } finally {
                this.setExecuting(false);
            }
        }
    }

    abstract protected void process(final T item);

    // *** items

    /**
     * Adds an item to this queue. Actions in the thread prior to calling this method happen-before
     * the passed argument is {@link #process(Object) processed}.
     * 
     * @param item the item to add.
     */
    public final void put(T item) {
        this.itemsLock.lock();
        try {
            this.items.add(item);
            this.notEmpty.signal();
        } finally {
            this.itemsLock.unlock();
        }
    }

    public final void eachItemDo(final IClosure<T> c) {
        this.itemsDo((items) -> {
            for (final T t : items) {
                c.executeChecked(t);
            }
            return null;
        });
    }

    /**
     * Allows <code>c</code> to arbitrarily modify our queue as it is locked during this method.
     * I.e. no items will be removed (passed to the closure) nor added.
     * 
     * @param c what to do with our queue.
     */
    public final void itemsDo(IClosure<? super Deque<T>> c) {
        this.itemsDo((q) -> {
            c.executeChecked(q);
            return null;
        });
    }

    public final <R> R itemsDo(Function<? super Deque<T>, R> c) {
        this.itemsLock.lock();
        try {
            final R res = c.apply(this.items);
            if (!this.items.isEmpty())
                this.notEmpty.signal();
            return res;
        } finally {
            this.itemsLock.unlock();
        }
    }

}