/*
* Created at 22:40 on 11/02/2017
*/
package com.example.news.analyser;

import com.example.news.Util;
import com.example.news.domain.News;

import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;

/**
 * @author zzhao
 */
class NewsAnalysingInspector implements Consumer<News>, Closeable {

    private final AtomicReference<InspectorContext> contextRef;

    private InspectorContext contextForExchange;

    private ScheduledExecutorService scheduler;

    private final ReentrantReadWriteLock.ReadLock readLock;

    private final ReentrantReadWriteLock.WriteLock writeLock;

    NewsAnalysingInspector(int numOfTopNewsToKeep) {
        this.contextRef = new AtomicReference<>(new InspectorContext(numOfTopNewsToKeep));
        this.contextForExchange = new InspectorContext(numOfTopNewsToKeep);
        final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
        this.readLock = lock.readLock();
        this.writeLock = lock.writeLock();
    }

    void start() {
        this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "inspector"));
        this.scheduler.scheduleAtFixedRate(this::inspect, 10, 10, TimeUnit.SECONDS);
    }

    void inspect() {
        // TODO find a clever way to avoid this locking: reference counting?
        final InspectorContext ctx = exchangeContext();
        if (ctx == null) {
            Util.error("failed inspecting");
            return;
        }

        Util.info("Positive news last 10s: %d", ctx.getNewsCounter().get());
        final News[] newsArray = toReverseSortedNewsArray(ctx.getPrioQueue());
        Arrays
                .stream(newsArray)
                .limit(ctx.getNumOfTopNewsTopKeep())
                .distinct()
                .forEach(news -> Util.info("Top prio news: %s", news));

        this.contextForExchange = ctx.reset();
    }

    private InspectorContext exchangeContext() {
        try {
            while (!this.writeLock.tryLock(1000, TimeUnit.MILLISECONDS)) {
                Util.warn("delayed inspecting");
            }
            try {
                return this.contextRef.getAndSet(this.contextForExchange);
            } finally {
                this.writeLock.unlock();
            }
        } catch (InterruptedException e) {
            Util.warn("inspector inspecting interrupted %s", e.getMessage());
            Thread.currentThread().interrupt();
            return null;
        }
    }

    static News[] toReverseSortedNewsArray(Collection<News> queue) {
        final News[] newsArray = queue.toArray(new News[queue.size()]);
        Arrays.sort(newsArray, Comparator.reverseOrder());
        return newsArray;
    }

    @Override
    public void accept(News news) {
        try {
            while (!this.readLock.tryLock(50, TimeUnit.MILLISECONDS)) {
                Util.warn("delayed accepting: %s", news);
            }
            try {
                this.contextRef.get().accept(news);
            } finally {
                this.readLock.unlock();
            }
        } catch (InterruptedException e) {
            Util.warn("inspector accepting interrupted %s", e.getMessage());
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void close() throws IOException {
        Util.shutdownAndAwaitTermination(this.scheduler, 5);
    }

    static class InspectorContext implements Consumer<News> {
        private final AtomicLong newsCounter;
        private final PriorityBlockingQueue<News> prioQueue;
        private final int numOfTopNewsTopKeep;

        InspectorContext(int topNewsToKeep) {
            this.numOfTopNewsTopKeep = topNewsToKeep;
            this.newsCounter = new AtomicLong(0);
            this.prioQueue = new PriorityBlockingQueue<>();
        }

        AtomicLong getNewsCounter() {
            return newsCounter;
        }

        PriorityBlockingQueue<News> getPrioQueue() {
            return prioQueue;
        }

        int getNumOfTopNewsTopKeep() {
            return numOfTopNewsTopKeep;
        }

        @Override
        public void accept(News news) {
            // not necessary to synchronize (count and queue) with analysing, results won't differ much
            this.newsCounter.incrementAndGet();
            // concurrent ops could result in more than 3 elements in the queue and possibly duplicate
            if (this.prioQueue.size() < this.numOfTopNewsTopKeep) { // parameterize number of top news
                this.prioQueue.offer(news);
            } else {
                if (news.compareTo(this.prioQueue.peek()) > 0) {
                    this.prioQueue.remove();
                    this.prioQueue.offer(news);
                }
            }
        }

        InspectorContext reset() {
            this.newsCounter.set(0);
            this.prioQueue.clear();
            return this;
        }
    }
}

@@ -550,24 +550,64 @@
 cReference;%0A
+import java.util.concurrent.locks.Lock;%0A
 import java.
@@ -687,16 +687,52 @@
 onsumer;
+%0Aimport java.util.function.Supplier;
 %0A%0A/**%0A *
@@ -2441,527 +2441,124 @@
     
-try %7B%0A            while (!this.writeLock.tryLock(1000, TimeUnit.MILLISECONDS)) %7B%0A                Util.warn(%22delayed inspecting%22);%0A            %7D%0A            try %7B%0A                return this.contextRef.getAndSet(this.contextForExchange);%0A            %7D finally %7B%0A                this.writeLock.unlock();%0A            %7D%0A        %7D catch (InterruptedException e) %7B%0A            Util.warn(%22inspector inspecting interrupted %25s%22, e.getMessage());%0A            Thread.currentThread().interrupt();%0A            return null;%0A        %7D
+return doWithinLock(this.writeLock, 1000,%0A                () -%3E this.contextRef.getAndSet(this.contextForExchange));
 %0A   
@@ -2844,35 +2844,69 @@
 news) %7B%0A        
-try
+doWithinLock(this.readLock, 50, () -%3E
  %7B%0A            w
@@ -2908,40 +2908,232 @@
     
-while (!this.readLock.tryLock(50
+this.contextRef.get().accept(news);%0A            return null;%0A        %7D);%0A    %7D%0A%0A    static %3CT%3E T doWithinLock(Lock lock, int timeoutMillis, Supplier%3CT%3E supplier) %7B%0A        try %7B%0A            while (!lock.tryLock(timeoutMillis
 , Ti
@@ -3187,36 +3187,31 @@
 rn(%22
-delayed accepting: %25s%22, news
+acquiring lock timeout%22
 );%0A 
@@ -3261,41 +3261,28 @@
     
-this.contextRef.get().accept(news
+return supplier.get(
 );%0A 
@@ -3320,26 +3320,17 @@
         
-this.readL
+l
 ock.unlo
@@ -3419,27 +3419,22 @@
 rn(%22
-inspector accepting
+acquiring lock
  int
@@ -3508,24 +3508,49 @@
 nterrupt();%0A
+            return null;%0A
         %7D%0A  

use lock util