/*
 * Copyright (C) 2016 VSCT
 *
 * 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 com.vsct.dt.strowgr.admin.nsq.consumer;

import fr.vsct.dt.nsq.NSQConfig;
import fr.vsct.dt.nsq.NSQConsumer;
import fr.vsct.dt.nsq.NSQMessage;
import fr.vsct.dt.nsq.lookup.NSQLookup;
import io.netty.channel.EventLoopGroup;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.disposables.Disposable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Optional;

public abstract class FlowableNSQConsumer<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(FlowableNSQConsumer.class);

    private final String topic;
    private final String channel;

    /* The flowable created by this consumer */
    private final Flowable<T> flowable;

    /*
     * We keep reference to consumer and to emitter to allow shutdown
     * and emission of onComplete. Make these reference volatile since they can be set in a different thread
     */
    private volatile NSQConsumer consumer;
    private volatile FlowableEmitter<NSQMessage> emitter;

    FlowableNSQConsumer(NSQLookup lookup, String topic, String channel, NSQConfig config) {
        this.topic = topic;
        this.channel = channel;

        Flowable<NSQMessage> flowable = Flowable
                .create(emitter -> {
                    this.emitter = emitter;
                    consumer = new NSQConsumer(lookup, topic, channel, emitter::onNext, config, emitter::onError);
                    consumer.setLookupPeriod(10 * 1000);
                    consumer.setMessagesPerBatch(10);

                    //We tell the NSQConsumer to use its own eventloop as the executor for message handling
                    //This prevent the use of the default cachedThreadPool, which spawns a huge amount of threads when a lot of messages arrives
                    //The only thing to take care of is that message transformation should never be blocking the event queue
                    //This has to be taken care of by child classes
                    EventLoopGroup eventLoopGroup = config.getEventLoopGroup();
                    consumer.setExecutor(eventLoopGroup);

                    consumer.start();

                    LOGGER.info("new subscription for FlowableNSQConsumer on topic {}, channel {}", topic, channel);

                    emitter.setDisposable(new Disposable() {
                        @Override
                        public void dispose() {
                            consumer.shutdown();
                        }

                        @Override
                        public boolean isDisposed() {
                            return eventLoopGroup.isShutdown();
                        }
                    });

                }, BackpressureStrategy.BUFFER);

        this.flowable = flowable
                .map(this::transformSafe)
                .filter(Optional::isPresent)
                .doOnError(throwable -> LOGGER.error("consumer error on topic " + topic, throwable))
                .map(Optional::get) //We don't want to stop the flowable on error
                .publish().autoConnect(); //Use publish/refCount for two reasons : NSQConsumer will be created only once. Observable will auto start
    }

    public Flowable<T> flowable() {
        return flowable;
    }

    private Optional<T> transformSafe(NSQMessage nsqMessage) {
        try {
            return Optional.ofNullable(transform(nsqMessage));
        } catch (Exception e) {
            LOGGER.error("can't deserialize the payload of message at {}, id={}, payload={}", nsqMessage.getTimestamp(), new String(nsqMessage.getId()), new String(nsqMessage.getMessage()), e);
            return Optional.empty();
        } finally {
            nsqMessage.finished();
        }
    }

    protected abstract T transform(NSQMessage nsqMessage) throws Exception;

    /**
     * Shutsdown this consumer and advice subscriber
     */
    public void shutdown() {
        LOGGER.info("stopping FlowableNSQConsumer on topic {}, channel {}", topic, channel);
        if (consumer != null) {
            consumer.shutdown();
        }
        if (emitter != null) {
            emitter.onComplete();
        }
    }

}

@@ -889,45 +889,8 @@
 le;%0A
-import io.reactivex.FlowableEmitter;%0A
 impo
@@ -987,16 +987,67 @@
 ctory;%0A%0A
+import java.util.ArrayList;%0Aimport java.util.List;%0A
 import j
@@ -1236,24 +1236,25 @@
 ring topic;%0A
+%0A
     private 
@@ -1282,16 +1282,24 @@
 %0A%0A    /*
+*%0A     *
  The flo
@@ -1328,16 +1328,21 @@
 consumer
+%0A    
  */%0A    
@@ -1384,24 +1384,25 @@
 %0A%0A    /*
+*
 %0A     * 
 We keep 
@@ -1397,287 +1397,133 @@
   * 
-We keep reference to consumer and to emitter to allow shutdown%0A     * and emission of onComplete. Make these reference volatile since they can be set in a different thread%0A     */%0A    private volatile NSQConsumer consumer;%0A    private volatile FlowableEmitter%3CNSQMessage%3E emitter
+Subscription's disposables references for shutdown%0A     */%0A    private final List%3CDisposable%3E disposables = new ArrayList%3C%3E()
 ;%0A%0A 
@@ -1686,29 +1686,13 @@
     
-Flowable%3CNSQMessage%3E 
+this.
 flow
@@ -1724,16 +1724,28 @@
        .
+%3CNSQMessage%3E
 create(e
@@ -1756,16 +1756,17 @@
 er -%3E %7B%0A
+%0A
         
@@ -1781,51 +1781,19 @@
     
-this.emitter = emitter;%0A                   
+NSQConsumer
  con
@@ -3211,43 +3211,8 @@
 FER)
-;%0A%0A        this.flowable = flowable
 %0A   
@@ -3508,130 +3508,182 @@
 sh()
-.autoConnect(); //Use publish/refCount for two reasons : NSQConsumer will be created only once. Observable will auto start
+ // do not start flowable immediately%0A                .autoConnect(1, disposables::add); // start flowable on first subscription and keep subscription's disposable references
 %0A   
@@ -4325,17 +4325,16 @@
   * Shut
-s
 down thi
@@ -4353,21 +4353,18 @@
 nd a
-dvice
+ll
  subscri
 ber%0A
@@ -4363,11 +4363,15 @@
 scri
-ber
+ptions.
 %0A   
@@ -4509,149 +4509,49 @@
     
-if (consumer != null) %7B%0A            consumer.shutdown();%0A        %7D%0A        if (emitter != null) %7B%0A            emitter.onComplete();%0A        %7D
+disposables.forEach(Disposable::dispose);
 %0A   

use Disposable to end FlowableNSQConsumer subscription's instead of using volatile emitter reference