/*
 * Copyright (c) 2018, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. 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.wso2.siddhi.core.util.event.handler;

import com.lmax.disruptor.EventHandler;
import org.apache.log4j.Logger;
import org.wso2.siddhi.core.event.Event;
import org.wso2.siddhi.core.stream.StreamJunction;

import java.util.LinkedList;
import java.util.List;

/**
 * Interface to be implemented to receive events via handlers.
 */
public class StreamHandler implements EventHandler<EventExchangeHolder> {

    private List<StreamJunction.Receiver> receivers;
    private int batchSize;
    private final String streamName;
    private final String siddhiAppName;
    private List<Event> eventBuffer = new LinkedList<>();
    private static final Logger log = Logger.getLogger(StreamHandler.class);


    public StreamHandler(List<StreamJunction.Receiver> receivers, int batchSize,
                         String streamName, String siddhiAppName) {
        this.receivers = receivers;
        this.batchSize = batchSize;
        this.streamName = streamName;
        this.siddhiAppName = siddhiAppName;
    }

    public void onEvent(EventExchangeHolder eventExchangeHolder, long sequence, boolean endOfBatch) {
        boolean isProcessed = eventExchangeHolder.getAndSetIsProcessed(true);
        if (!isProcessed) {
            eventBuffer.add(eventExchangeHolder.getEvent());
            if (eventBuffer.size() == batchSize || endOfBatch) {
                for (StreamJunction.Receiver receiver : receivers) {
                    try {
                        receiver.receive(eventBuffer);
                    } catch (Throwable t) {
                        log.error("Error in SiddhiApp '" + siddhiAppName +
                                "' after consuming events from Stream " +
                                "'" + streamName + "', " + t.getMessage(), t);
                    }
                }
                eventBuffer.clear();
            }
        } else if (endOfBatch) {
            if (eventBuffer.size() != 0) {
                for (StreamJunction.Receiver receiver : receivers) {
                    try {
                        receiver.receive(eventBuffer);
                    } catch (Throwable t) {
                        log.error("Error in SiddhiApp '" + siddhiAppName +
                                "' after consuming events from Stream " +
                                "'" + streamName + "', " + t.getMessage(), t);
                    }
                }
                eventBuffer.clear();
            }
        }

    }
}

@@ -1370,16 +1370,127 @@
 class);%0A
+    private final StreamJunction faultStreamJunction;%0A    private final StreamJunction.FaultAction faultAction;
 %0A%0A    pu
@@ -1628,16 +1628,117 @@
 iAppName
+, StreamJunction faultStreamJunction,%0A                         StreamJunction.FaultAction faultAction
 ) %7B%0A    
@@ -1883,24 +1883,120 @@
 dhiAppName;%0A
+        this.faultStreamJunction = faultStreamJunction;%0A        this.faultAction = faultAction;%0A
     %7D%0A%0A    p
@@ -2546,206 +2546,27 @@
     
-log.error(%22Error in SiddhiApp '%22 + siddhiAppName +%0A                                %22' after consuming events from Stream %22 +%0A                                %22'%22 + streamName + %22', %22 + t.getMessage()
+onError(eventBuffer
 , t)
@@ -2924,32 +2924,354 @@
 owable t) %7B%0A    
+                    onError(eventBuffer, t);%0A                    %7D%0A                %7D%0A                eventBuffer.clear();%0A            %7D%0A        %7D%0A%0A    %7D%0A%0A    private void onError(List%3CEvent%3E eventBuffer, Throwable t) %7B%0A        switch (faultAction) %7B%0A            case LOG:%0A                for (Event event : eventBuffer) %7B%0A
                 
@@ -3353,20 +3353,16 @@
         
-    
 %22' after
@@ -3423,20 +3423,16 @@
         
-    
 %22'%22 + st
@@ -3468,91 +3468,781 @@
 ge()
-, t);%0A                    %7D%0A                %7D%0A                eventBuffer.clear();%0A
+ + %22. Hence, dropping event '%22%0A                            + event.toString() + %22'%22, t);%0A                %7D%0A                break;%0A            case STREAM:%0A                for (Event event : eventBuffer) %7B%0A                    if (faultStreamJunction != null) %7B%0A                        faultStreamJunction.sendEvent(event);%0A                    %7D else %7B%0A                        log.error(%22Error in SiddhiApp '%22 + siddhiAppName +%0A                                %22' after consuming events from Stream %22 +%0A                                %22'%22 + streamName + %22', %22 + t.getMessage()%0A                                + %22. Siddhi Fault Stream for '%22 + streamName + %22' is not defined. %22%0A                                + %22Hence dropping the event '%22 + event.toString() + %22'%22, t);%0A      
     
@@ -4245,33 +4245,42 @@
             
-%7D%0A
+  %7D%0A       
         
 %7D%0A%0A    %7D%0A%7D%0A
@@ -4264,23 +4264,100 @@
             
+ 
 %7D%0A
+                break;%0A            default:%0A                break;%0A        %7D
 %0A    %7D%0A%7D%0A

Handle errors while buffering events with Async annotation