/* 
 * Copyright 2014 Informatica Corp.
 *
 * 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.informatica.surf;

import com.informatica.um.binge.api.impl.VDSEventImpl;
import com.informatica.um.binge.api.impl.VDSEventListImpl;
import com.informatica.um.binge.api.impl.VDSEventListPoolFactory;
import com.informatica.um.binge.api.impl.VDSEventPoolFactory;
import com.informatica.um.binge.api.impl.VDSMessageAckSource;
import com.informatica.vds.api.*;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.apache.commons.pool.impl.GenericKeyedObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 *
 * @author jraj
 */
public class Node implements Runnable, EventHandler<SurfEvent> {
    
    private final VDSSource _source;
    private final VDSTarget _target;
    private final Map<Object, VDSConfiguration> _context;
    private static final Logger _logger = LoggerFactory.getLogger(Node.class);
    private volatile boolean _shutdown = false;
    private final VDSMessageAckSource _acksource;
    private final boolean _needsAck;
    private final GenericKeyedObjectPool <Integer, VDSEventListImpl>_eventListPool;
    private final List<VDSTransform> _transforms;
    
    public Node(VDSSource source, VDSTarget target, List<VDSTransform> transforms, Map<Object, VDSConfiguration> ctx){
        _logger.info("Creating node...");
        _source = source;
        _target = target;
        _transforms = transforms;
        _context = ctx;
        if(source instanceof VDSMessageAckSource){
            _acksource = (VDSMessageAckSource)source;
            _needsAck = true;
        }
        else{
            _acksource = null;
            _needsAck = false;
        }
        GenericKeyedObjectPool<Integer, VDSEventImpl> evtpool = new GenericKeyedObjectPool(new VDSEventPoolFactory());
        evtpool.setMaxActive(-1);
        evtpool.setMaxTotal(1000);
        _eventListPool = new GenericKeyedObjectPool<>(new VDSEventListPoolFactory(evtpool));
        _eventListPool.setMaxActive(-1);
        _eventListPool.setMaxTotal(1000);
    }
    
    public void open() throws Exception{
        _logger.info("Opening source and target connections...");
        _source.open(_context.get(_source));
        _target.open(_context.get(_target));
        _logger.info("Initializing transforms");
        for(VDSTransform tx: _transforms){
            tx.open(_context.get(tx));
        }
    }
    
    @Override
    public void run(){
        _logger.info("Node run starting...");
        Executor executor = Executors.newCachedThreadPool();
        Disruptor<SurfEvent> disruptor = new Disruptor<>(SurfEvent.EVENT_FACTORY, 128, executor);
        disruptor.handleEventsWith(this);
        RingBuffer<SurfEvent> buffer = disruptor.start();


        VDSEventListImpl events = null;
        while(!_shutdown){
            try{
                events = _eventListPool.borrowObject(0);
                _source.read(events);
                if(_needsAck){
                    Object obj = _acksource.getInputObject();
                    // Hacking this in for now
                    // Need to actually ack the object after Kinesis consumes it: TODO
                    _acksource.updateInputObject(obj);
                }
                long seq = buffer.next();
                SurfEvent event = buffer.get(seq);
                event.setEventlist(events);
                buffer.publish(seq);
            }
            catch(Exception ex){
                _logger.error("Exception while reading data:", ex);
                ex.printStackTrace();
            }
        }
    }
    
    public void shutdown(){
        _shutdown = true;
    }

    @Override
    public void onEvent(SurfEvent surfEvent, long l, boolean b) throws Exception {
        VDSEventListImpl srcEvents = surfEvent.getEventlist();
        VDSEventListImpl txEvents = applyTransforms(srcEvents, _transforms.iterator());
        _logger.debug("Sending events to target...");
        for(VDSEventImpl evt: txEvents.getEventsList()){
            _target.write(evt);
        }
        _eventListPool.returnObject(0, txEvents);
    }

    private VDSEventListImpl applyTransforms(VDSEventListImpl srcEvents, Iterator<VDSTransform> transforms) throws Exception{
        if(!transforms.hasNext()){
            return srcEvents;
        }
        VDSTransform tx = transforms.next();
        VDSEventListImpl outlist = _eventListPool.borrowObject(0);
        for(VDSEventImpl evt: srcEvents.getEventsList()){
            tx.apply(evt, outlist);
        }
        _eventListPool.returnObject(0, srcEvents);
        return applyTransforms(outlist, transforms);

    }

}

@@ -2617,24 +2617,25 @@
 axTotal(1000
+0
 );%0A        _
@@ -2798,16 +2798,17 @@
 tal(1000
+0
 );%0A    %7D
@@ -3639,32 +3639,94 @@
 orrowObject(0);%0A
+                while(events.getEventsList().isEmpty()) %7B%0A    
                 
@@ -3739,32 +3739,50 @@
 e.read(events);%0A
+                %7D%0A
                 
@@ -4683,16 +4683,62 @@
 Events =
+ null;%0A        try%7B%0A                txEvents =
  applyTr
@@ -4778,24 +4778,192 @@
 terator());%0A
+        %7D%0A        catch(Exception ex)%7B%0A            _logger.warn(%22Exception while invoking transforms. Skipping event.%22, ex);%0A            return;%0A        %7D%0A        try%7B%0A
         _log
@@ -5095,32 +5095,166 @@
 evt);%0A        %7D%0A
+        %7D%0A        catch(Exception ex)%7B%0A            _logger.warn(%22Exception while handling event%22, ex);%0A        %7D%0A        finally%7B%0A    
         _eventLi
@@ -5283,24 +5283,34 @@
  txEvents);%0A
+        %7D%0A
     %7D%0A%0A    p

Improve exception handling