/*
 *  @author Ranides Atterwim <ranides@gmail.com>
 *  @copyright Ranides Atterwim
 *  @license WTFPL
 *  @url http://ranides.net/projects/assira
 */

package net.ranides.assira.events;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import net.ranides.assira.trace.AdvLogger;
import net.ranides.assira.trace.LoggerUtils;

/**
 * Klasa bazowa dla implementacji zwracanych przez metody 
 * {@link EventLock#lock},
 * {@link EventLock#weakLock},
 * {@link EventLock#singleLock},
 * @author ranides
 */
public abstract class EventLockQue extends EventLock implements EventListener<Event> {
    
    private static final AdvLogger LOGGER = LoggerUtils.getLogger();
   
    private final BlockingQueue<Event> events;
    private int discarded;
    private int processed;
    
/* ************************************************************************** */
    
    EventLockQue() {
        this.processed = 0;
        this.discarded = 0;
        this.events = new LinkedBlockingQueue<>();
    }
    
    EventLockQue(int capacity) {
        this.processed = 0;
        this.discarded = 0;
        this.events = new LinkedBlockingQueue<>(capacity);
    }
    
    @Override
    public Event waitForEvent() throws InterruptedException {
        return events.take();
    }
    
    @Override
    public Event waitForEvent(long timeout) throws InterruptedException {
        return events.poll(timeout, TimeUnit.MILLISECONDS);
    }
  
    @Override
    public <T extends Event> T waitForEvent(Class<T> event) throws InterruptedException {
        while(true) {
            Event recent = events.take();
            if( event.isInstance(recent) ) {
                return event.cast(recent);
            }
        }
    }
    
    @Override
    public <T extends Event> T waitForEvent(Class<T> event, long timeout) throws InterruptedException {
        long end = System.currentTimeMillis() + timeout;
        long rem = timeout;
        while(rem > 0) {
            Event recent = events.poll(rem, TimeUnit.MILLISECONDS);
            if( event.isInstance(recent) ) {
                return event.cast(recent);
            }
            rem = end - System.currentTimeMillis();
        }
        return null;
    }
    
    @Override
    public void release() {
        LOGGER.ftrace("overall processed=%d", processed);
        if( discarded > 0 ) {
            LOGGER.fwarn("overall discarded=%d", discarded);
        }
    }
    
    @Override
    public void reset() {
        events.clear();
    }
    
    @Override
    public EventListener<Event> listener() {
        return this;
    }
    
    @Override
    public void handleEvent(Event event) {
        processed++;
        while( !events.offer(event) ) { 
            Event ignored = events.poll();
            discarded++;
            if( 0 == (processed % 100) ) {
                LOGGER.fwarn("events flood. processed=%d discarded=%d event=%s", processed, discarded, ignored);
            }
        }
    }

    /**
     * Zwraca liczbę komunikatów zgubionych przez cały czas istnienia obiektu.
     * @return 
     */
    public int discarded() {
        return discarded;
    }

    /**
     * Zwraca liczbę wszystkich otrzymanych komunikatów przez cały czas istnienia obiektu
     * (w tym komuniaktów zgubionych).
     * @return 
     */
    public int counter() {
        return processed;
    }
    
    /**
     * Zwraca liczbę aktualnie oczekujących komunikatów.
     * @return 
     */
    public int pending() {
        return events.size();
    }

/* ************************************************************************** */
    
    /**
     * Implementacja zwracana przez metodę {@link EventLock#singleLock}
     */
    static class SingleLock extends OwnableLockQue {
        
        public <T extends Event> SingleLock(Class<T> event, EventRouter router) {
            super(event, router);
        }

        public <T extends Event> SingleLock(Class<T> event, EventRouter router, int capacity) {
            super(event, router, capacity);
        }
        
        @Override
        public Event waitForEvent() throws InterruptedException {
            try {
                return super.waitForEvent();
            } finally {
                release();
            }
        }

        @Override
        public <T extends Event> T waitForEvent(Class<T> event) throws InterruptedException {
            try {
                return super.waitForEvent(event);
            } finally {
                release();
            }
        }
        
    }
    
    /**
     * Implementacja zwracana przez metody {@link EventLock#lock}, {@link EventLock#weakLock}.
     */
    static class OwnableLockQue extends EventLockQue {
        
        private final Class<? extends Event> observed;
        private final EventRouter router;
        
        public OwnableLockQue(Class<? extends Event> event, EventRouter router) {
            super();
            this.observed = event;
            this.router = router;
        }

        public OwnableLockQue(Class<? extends Event> event, EventRouter router, int capacity) {
            super(capacity);
            this.observed = event;
            this.router = router;
        }
        
        @Override
        public void release() {
            super.release();
            router.removeEventListener(observed, this);
        }
    }
    
}
