@Override
    public void onEvent(Object event, long sequence, boolean endOfBatch) throws Exception {
        try {
            if (event == null) {
                return;
            }
            Runnable runnable = null;
            if (event instanceof Tuple) {
                if (((TupleExt) event).isBatchTuple()) {
                    List<Object> values = ((Tuple) event).getValues();
                    for (Object value : values) {
                        Pair<MessageId, List<Object>> val = (Pair<MessageId, List<Object>>) value;
                        TupleImplExt tuple = new TupleImplExt(
                                sysTopologyCtx, val.getSecond(), val.getFirst(), ((TupleImplExt) event));
                        runnable = processTupleEvent(tuple);
                        if (runnable != null) {
                            runnable.run();
                            runnable = null;
                        }
                    }
                } else {
                    runnable = processTupleEvent((Tuple) event);
                }
            } else if (event instanceof TimerTrigger.TimerEvent) {
                processTimerEvent((TimerTrigger.TimerEvent) event);
                return;
            } else if (event instanceof IAckMsg) {
                runnable = (Runnable) event;
            } else if (event instanceof Runnable) {
                runnable = (Runnable) event;
            } else {

                LOG.warn("Receive one unknown event-" + event.toString() + " " + idStr);
                return;
            }

            if (runnable != null)
                runnable.run();

        } catch (Throwable e) {
            if (!taskStatus.isShutdown()) {
                LOG.info("Unknown exception ", e);
                reportError.report(e);
            }
        }
    }
@ Override public void onEvent ( Object event , long sequence , boolean endOfBatch ) throws Exception { try { if ( event == null ) { return ; } Runnable runnable = null ; if ( event instanceof Tuple ) { if ( ( ( TupleExt ) event ) . isBatchTuple ( ) ) { List < Object > values = ( ( Tuple ) event ) . getValues ( ) ; for ( Object value : values ) { Pair < MessageId , List < Object > > val = ( Pair < MessageId , List < Object > > ) value ; TupleImplExt tuple = new TupleImplExt ( sysTopologyCtx , val . getSecond ( ) , val . getFirst ( ) , ( ( TupleImplExt ) event ) ) ; runnable = processTupleEvent ( tuple ) ; if ( runnable != null ) { runnable . run ( ) ; runnable = null ; } } } else { runnable = processTupleEvent ( ( Tuple ) event ) ; } } else if ( event instanceof TimerTrigger . TimerEvent ) { processTimerEvent ( ( TimerTrigger . TimerEvent ) event ) ; return ; } else if ( event instanceof IAckMsg ) { runnable = ( Runnable ) event ; } else if ( event instanceof Runnable ) { runnable = ( Runnable ) event ; } else { LOG . warn ( "Receive one unknown event-" + event . toString ( ) + " " + idStr ) ; return ; } if ( runnable != null ) runnable . run ( ) ; } catch ( Throwable e ) { if ( ! taskStatus . isShutdown ( ) ) { LOG . info ( "Unknown exception " , e ) ; reportError . report ( e ) ; } } }
Handle acker message

@see com.lmax.disruptor.EventHandler#onEvent(java.lang.Object, long, boolean)