/**
 * Copyright 2016 AWACS Project.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
package io.awacs.agent;

import com.google.common.collect.Lists;
import io.awacs.core.transport.Message;
import io.awacs.core.transport.ResponseHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Enable batch send
 * Created by pixyonly on 4/21/17.
 */
final class MessageAccumulator {

    private boolean batchInProgress = false;

    private List<Message> batch;

    private HashedWheelTimer timer;

    private AtomicLong chunkBytes = new AtomicLong(0l);

    private ConcurrentMap<String, ResultTracker> trackers;

    private int maxBatchSize;

    private long maxChunkBytes;

    private long lingerMs;

    private long timeout;

    private NettyClient client;

    private ExecutorService executorService;

    MessageAccumulator(NettyClient client,
                       int maxBatchSize,
                       long maxChunkBytes,
                       long lingerms,
                       long timeout) {
        this.client = client;
        this.batch = new ArrayList<>(maxBatchSize);
        this.maxBatchSize = maxBatchSize;
        this.maxChunkBytes = maxChunkBytes;
        this.lingerMs = lingerms;
        this.timeout = timeout;
        this.timer = new HashedWheelTimer();
        executorService = Executors.newSingleThreadExecutor();
    }

    void flush(final Collection<Message> ready, final byte[] buf) {
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                int offset = 0;
                for (Message m : ready) {
                    System.arraycopy(m.serialize(), 0, buf, offset, m.size());
                    offset += m.size();
                }
                client.send(buf);
            }
        });

    }

    synchronized void append(Message message, final ResponseHandler<? extends Message> callback) {
        if (!batchInProgress) {
            batchInProgress = true;
            this.timer.newTimeout(new TimerTask() {
                @Override
                public void run(Timeout timeout) throws Exception {
                    final List<Message> ready = Lists.newArrayList();
                    ready.addAll(batch);
                    final byte[] buf = new byte[chunkBytes.intValue()];
                    flush(ready, buf);
                    batch.clear();
                    chunkBytes.set(0);
                    batchInProgress = false;
                }
            }, lingerMs, TimeUnit.MILLISECONDS);
        }
        batch.add(message);
        chunkBytes.addAndGet(message.size());
        if (batch.size() >= maxBatchSize || chunkBytes.get() >= maxChunkBytes) {
            final List<Message> ready = Lists.newArrayList();
            ready.addAll(batch);
            final byte[] buf = new byte[chunkBytes.intValue()];
            flush(ready, buf);
            batch.clear();
            chunkBytes.set(0);
        }
    }

    //TODO
    void confirm(String rid) {
    }

    synchronized void close() {
        try {
            final List<Message> ready = Lists.newArrayList();
            ready.addAll(batch);
            final byte[] buf = new byte[chunkBytes.intValue()];
            flush(ready, buf);
            executorService.shutdown();
            executorService.awaitTermination(lingerMs + 1000, TimeUnit.MILLISECONDS);
            timer.stop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class ResultTracker {

    }

}

@@ -1089,16 +1089,66 @@
 meUnit;%0A
+import java.util.concurrent.atomic.AtomicBoolean;%0A
 import j
@@ -1297,17 +1297,23 @@
 private 
-b
+AtomicB
 oolean b
@@ -1329,21 +1329,40 @@
 gress = 
+new AtomicBoolean(
 false
+)
 ;%0A%0A    p
@@ -1740,16 +1740,58 @@
 Service;
+%0A    %0A    private Timeout batchStandingBy;
 %0A%0A    Me
@@ -2913,16 +2913,22 @@
 Progress
+.get()
 ) %7B%0A    
@@ -2954,15 +2954,18 @@
 ress
- = 
+.set(
 true
+)
 ;%0A  
@@ -2970,29 +2970,42 @@
             
-this.
+batchStandingBy = 
 timer.newTim
@@ -3121,16 +3121,80 @@
 ption %7B%0A
+                    synchronized(MessageAccumulator.this) %7B%0A    
         
@@ -3247,32 +3247,36 @@
 newArrayList();%0A
+    
                 
@@ -3312,32 +3312,36 @@
                 
+    
 final byte%5B%5D buf
@@ -3388,32 +3388,36 @@
                 
+ 
+   
 flush(ready, buf
@@ -3431,32 +3431,36 @@
                 
+    
 batch.clear();%0A 
@@ -3470,32 +3470,36 @@
                 
+  
+  
 chunkBytes.set(0
@@ -3513,32 +3513,36 @@
                 
+    
 batchInProgress 
@@ -3544,17 +3544,42 @@
 ress
- = false;
+.set(false);%0A                    %7D
 %0A   
@@ -4047,32 +4047,110 @@
 nkBytes.set(0);%0A
+            batchInProgress.set(false);%0A            batchStandingBy.cancel();%0A
         %7D%0A    %7D%0A

Update MessageAccumulator.java