/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.usergrid.services.notifications;

import org.apache.usergrid.metrics.MetricsFactory;
import org.apache.usergrid.mq.*;
import org.apache.usergrid.persistence.EntityManager;
import org.apache.usergrid.persistence.EntityManagerFactory;
import org.apache.usergrid.services.ServiceManager;
import org.apache.usergrid.services.ServiceManagerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.observables.GroupedObservable;
import rx.schedulers.Schedulers;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

public class QueueListener  {
    public static int MAX_CONSECUTIVE_FAILS = 10;


    private static final Logger LOG = LoggerFactory.getLogger(QueueListener.class);

    @Autowired
    private MetricsFactory metricsService;

    @Autowired
    private ServiceManagerFactory smf;

    @Autowired
    private EntityManagerFactory emf;

    @Autowired
    private Properties properties;

    private org.apache.usergrid.mq.QueueManager queueManager;

    private ServiceManager svcMgr;

    private long sleepWhenNoneFound = 0;

    private long sleepBetweenRuns = 5000;

    ExecutorService pool;
    List<Future> futures;

    public static final String MAX_THREADS = "1";

    public QueueListener() {
        pool = Executors.newFixedThreadPool(1);
    }
    public QueueListener(ServiceManagerFactory smf, EntityManagerFactory emf, MetricsFactory metricsService, Properties props){
        this();
        this.smf = smf;
        this.emf = emf;
        this.metricsService = metricsService;
        this.properties = props;
    }

    @PostConstruct
    void init() {
        run();
    }

    public void run(){
        boolean shouldRun = new Boolean(properties.getProperty("usergrid.notifications.listener.run", "true"));

        if(shouldRun) {
            LOG.info("QueueListener: starting.");
            int threadCount = 0;

            try {
                sleepBetweenRuns = new Long(properties.getProperty("usergrid.notifications.listener.sleep.between", "0")).longValue();
                sleepWhenNoneFound = new Long(properties.getProperty("usergrid.notifications.listener.sleep.after", "5000")).longValue();
                int maxThreads = new Integer(properties.getProperty("usergrid.notifications.listener.maxThreads", MAX_THREADS));
                futures = new ArrayList<Future>(maxThreads);
                while (threadCount++ < maxThreads) {
                    LOG.info("QueueListener: Starting thread {}.", threadCount);
                    futures.add(
                            pool.submit(new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        execute();
                                    } catch (Exception e) {
                                        LOG.error("failed to start push", e);
                                    }
                                }
                            })
                    );
                }
            } catch (Exception e) {
                LOG.error("QueueListener: failed to start:", e);
            }
            LOG.info("QueueListener: done starting.");
        }else{
            LOG.info("QueueListener: never started due to config value usergrid.notifications.listener.run.");
        }

    }

    private void execute(){

        svcMgr = smf.getServiceManager(smf.getManagementAppId());
        queueManager = svcMgr.getQueueManager();
        final AtomicInteger consecutiveExceptions = new AtomicInteger();
        LOG.info("QueueListener: Starting execute process.");

        // run until there are no more active jobs
        while ( true ) {
            try {
                QueueResults results = ApplicationQueueManager.getDeliveryBatch(queueManager);
                LOG.info("QueueListener: retrieved batch of {} messages",results.size());

                List<Message> messages = results.getMessages();
                if(messages.size()>0) {
                    HashMap<UUID,List<ApplicationQueueMessage>> messageMap = new HashMap<>(messages.size());
                    //group messages into hash map by app id
                    for(Message message : messages){
                        ApplicationQueueMessage queueMessage = ApplicationQueueMessage.generate(message);
                        UUID applicationId = queueMessage.getApplicationId();
                        if(!messageMap.containsKey(applicationId)){
                            List<ApplicationQueueMessage> applicationQueueMessages = new ArrayList<ApplicationQueueMessage>();
                            applicationQueueMessages.add(queueMessage);
                            messageMap.put(applicationId,applicationQueueMessages);
                        }else{
                            messageMap.get(applicationId).add(queueMessage);
                        }
                    }
                    //send each set of app ids together
                    for(Map.Entry<UUID,List<ApplicationQueueMessage>> entry : messageMap.entrySet()){
                        UUID applicationId = entry.getKey();
                        EntityManager entityManager = emf.getEntityManager(applicationId);
                        ServiceManager serviceManager = smf.getServiceManager(applicationId);
                        final ApplicationQueueManager manager = new ApplicationQueueManager(
                                new JobScheduler(serviceManager, entityManager),
                                entityManager,
                                queueManager,
                                metricsService,
                                properties
                        );
                        long now = System.currentTimeMillis();
                        LOG.info("QueueListener: send batch {} messages", entry.getValue().size());
                        manager.sendBatchToProviders(entry.getValue()).toBlocking().lastOrDefault(null);
                        LOG.info("QueueListener: sent batch {} messages duration {} ms", entry.getValue().size(),System.currentTimeMillis() - now);
                    }

                    if(sleepBetweenRuns > 0) {
                        Thread.sleep(sleepBetweenRuns);
                    }
                }
                else{
                    LOG.info("QueueListener: no messages...sleep...", results.size());
                    Thread.sleep(sleepWhenNoneFound);
                }
                //send to the providers
                consecutiveExceptions.set(0);
            }catch (Exception ex){
                LOG.error("failed to dequeue",ex);
                if(consecutiveExceptions.getAndIncrement() > MAX_CONSECUTIVE_FAILS){
                    LOG.error("killing message listener; too many failures");
                    break;
                }
            }
        }
    }

    public void stop(){
        LOG.info("QueueListener: stop processes");

        for(Future future : futures){
            future.cancel(true);
        }
    }



}

@@ -6058,32 +6058,136 @@
               %7D%0A
+                    long now = System.currentTimeMillis();%0A                    Observable merge = null;%0A
                 
@@ -6963,70 +6963,8 @@
  );%0A
-                        long now = System.currentTimeMillis();
 %0A   
@@ -7016,24 +7016,38 @@
  send batch 
+for app %7B%7D of 
 %7B%7D messages%22
@@ -7047,16 +7047,32 @@
 ssages%22,
+ entry.getKey(),
  entry.g
@@ -7113,16 +7113,37 @@
         
+ Observable current =
  manager
@@ -7181,16 +7181,278 @@
 Value())
+;%0A                        if(merge == null)%0A                            merge = current;%0A                        else %7B%0A                            merge = Observable.merge(merge,current);%0A                        %7D%0A                    %7D%0A                    merge
 .toBlock
@@ -7478,20 +7478,16 @@
 (null);%0A
-    
         
@@ -7563,32 +7563,24 @@
 %7B%7D ms%22, 
-entry.getValue()
+messages
 .size(),
@@ -7613,38 +7613,16 @@
  - now);
-%0A                    %7D
 %0A%0A      

add rx merge code