@Override public synchronized void acceptInstance(InstanceEvent e){
  m_busy=true;
  if (m_completed == null) {
    m_completed=new HashMap<Object,Instances>();
    m_incrementalSavers=new HashMap<Object,ArffSaver>();
    m_finishedCount=0;
    m_incrementalCounter=0;
  }
  if (e.getStatus() == InstanceEvent.FORMAT_AVAILABLE) {
    if (m_completed.containsKey(e.getSource())) {
      if (m_log != null) {
        String msg=statusMessagePrefix() + "Resetting appender.";
        m_log.statusMessage(msg);
        m_log.logMessage("[Appender] " + msg + " New start of stream detected before "+ "all incoming streams have finished!");
      }
      m_completed=new HashMap<Object,Instances>();
      m_incrementalSavers=new HashMap<Object,ArffSaver>();
      m_incrementalCounter=0;
      m_completeHeader=null;
      m_finishedCount=0;
    }
    m_completed.put(e.getSource(),e.getStructure());
    if (m_completed.size() == m_listenees.size()) {
      try {
        if (m_log != null) {
          String msg=statusMessagePrefix() + "Making output header";
          m_log.statusMessage(msg);
          m_log.logMessage("[Appender] " + msg);
        }
        m_completeHeader=makeOutputHeader();
        m_ie.setStructure(m_completeHeader);
        notifyInstanceListeners(m_ie);
        if (m_incrementalSavers.size() > 0) {
          for (          ArffSaver s : m_incrementalSavers.values()) {
            s.writeIncremental(null);
            File tmpFile=s.retrieveFile();
            ArffLoader loader=new ArffLoader();
            loader.setFile(tmpFile);
            Instances tempStructure=loader.getStructure();
            Instance tempLoaded=loader.getNextInstance(tempStructure);
            while (tempLoaded != null) {
              Instance converted=makeOutputInstance(m_completeHeader,tempLoaded);
              m_ie.setStatus(InstanceEvent.INSTANCE_AVAILABLE);
              m_ie.setInstance(converted);
              notifyInstanceListeners(m_ie);
              m_incrementalCounter++;
              if (m_incrementalCounter % 10000 == 0) {
                if (m_log != null) {
                  m_log.statusMessage(statusMessagePrefix() + "Processed " + m_incrementalCounter+ " instances");
                }
              }
              tempLoaded=loader.getNextInstance(tempStructure);
            }
          }
          m_incrementalSavers.clear();
        }
      }
 catch (      Exception e1) {
        String msg=statusMessagePrefix() + "ERROR: unable to create output instances structure.";
        if (m_log != null) {
          m_log.statusMessage(msg);
          m_log.logMessage("[Appender] " + e1.getMessage());
        }
        stop();
        e1.printStackTrace();
        m_busy=false;
        return;
      }
    }
    m_busy=false;
    return;
  }
  if (e.getStatus() == InstanceEvent.BATCH_FINISHED || e.getStatus() == InstanceEvent.INSTANCE_AVAILABLE) {
    Instance currentI=e.getInstance();
    if (m_completeHeader == null) {
      if (currentI != null) {
        ArffSaver saver=m_incrementalSavers.get(e.getSource());
        if (saver == null) {
          saver=new ArffSaver();
          try {
            File tmpFile=File.createTempFile("weka",".arff");
            saver.setFile(tmpFile);
            saver.setRetrieval(weka.core.converters.Saver.INCREMENTAL);
            saver.setInstances(new Instances(currentI.dataset(),0));
            m_incrementalSavers.put(e.getSource(),saver);
          }
 catch (          IOException e1) {
            stop();
            e1.printStackTrace();
            String msg=statusMessagePrefix() + "ERROR: unable to save instance to temp file";
            if (m_log != null) {
              m_log.statusMessage(msg);
              m_log.logMessage("[Appender] " + e1.getMessage());
            }
            m_busy=false;
            return;
          }
        }
        try {
          saver.writeIncremental(currentI);
          if (e.getStatus() == InstanceEvent.BATCH_FINISHED) {
            m_finishedCount++;
          }
        }
 catch (        IOException e1) {
          stop();
          e1.printStackTrace();
          String msg=statusMessagePrefix() + "ERROR: unable to save instance to temp file";
          if (m_log != null) {
            m_log.statusMessage(msg);
            m_log.logMessage("[Appender] " + e1.getMessage());
          }
          m_busy=false;
          return;
        }
      }
    }
 else {
      if (currentI != null) {
        int code=InstanceEvent.INSTANCE_AVAILABLE;
        if (e.getStatus() == InstanceEvent.BATCH_FINISHED) {
          m_finishedCount++;
          if (m_finishedCount == m_listenees.size()) {
            code=InstanceEvent.BATCH_FINISHED;
          }
        }
        Instance newI=makeOutputInstance(m_completeHeader,currentI);
        m_ie.setStatus(code);
        m_ie.setInstance(newI);
        notifyInstanceListeners(m_ie);
        m_incrementalCounter++;
        if (m_incrementalCounter % 10000 == 0) {
          if (m_log != null) {
            m_log.statusMessage(statusMessagePrefix() + "Processed " + m_incrementalCounter+ " instances");
          }
        }
        if (code == InstanceEvent.BATCH_FINISHED) {
          if (m_log != null) {
            m_log.statusMessage(statusMessagePrefix() + "Finished");
          }
          m_completed=null;
          m_incrementalSavers=null;
          m_incrementalCounter=0;
          m_completeHeader=null;
          m_finishedCount=0;
        }
      }
    }
  }
  m_busy=false;
}

@ Override public synchronized void acceptInstance ( InstanceEvent e ) { m_busy = true ; if ( m_completed == null ) { m_completed = new HashMap < Object , Instances > ( ) ; m_incrementalSavers = new HashMap < Object , ArffSaver > ( ) ; m_finishedCount = 0 ; m_incrementalCounter = 0 ; } if ( e . getStatus ( ) == InstanceEvent . FORMAT_AVAILABLE ) { if ( m_completed . containsKey ( e . getSource ( ) ) ) { if ( m_log != null ) { String msg = statusMessagePrefix ( ) + "Resetting appender." ; m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + msg + " New start of stream detected before " + "all incoming streams have finished!" ) ; } m_completed = new HashMap < Object , Instances > ( ) ; m_incrementalSavers = new HashMap < Object , ArffSaver > ( ) ; m_incrementalCounter = 0 ; m_completeHeader = null ; m_finishedCount = 0 ; } m_completed . put ( e . getSource ( ) , e . getStructure ( ) ) ; if ( m_completed . size ( ) == m_listenees . size ( ) ) { try { if ( m_log != null ) { String msg = statusMessagePrefix ( ) + "Making output header" ; m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + msg ) ; } m_completeHeader = makeOutputHeader ( ) ; m_ie . setStructure ( m_completeHeader ) ; notifyInstanceListeners ( m_ie ) ; if ( m_incrementalSavers . size ( ) > 0 ) { for ( ArffSaver s : m_incrementalSavers . values ( ) ) { s . writeIncremental ( null ) ; File tmpFile = s . retrieveFile ( ) ; ArffLoader loader = new ArffLoader ( ) ; loader . setFile ( tmpFile ) ; Instances tempStructure = loader . getStructure ( ) ; Instance tempLoaded = loader . getNextInstance ( tempStructure ) ; while ( tempLoaded != null ) { Instance converted = makeOutputInstance ( m_completeHeader , tempLoaded ) ; m_ie . setStatus ( InstanceEvent . INSTANCE_AVAILABLE ) ; m_ie . setInstance ( converted ) ; notifyInstanceListeners ( m_ie ) ; m_incrementalCounter ++ ; if ( m_incrementalCounter % 10000 == 0 ) { if ( m_log != null ) { m_log . statusMessage ( statusMessagePrefix ( ) + "Processed " + m_incrementalCounter + " instances" ) ; } } tempLoaded = loader . getNextInstance ( tempStructure ) ; } } m_incrementalSavers . clear ( ) ; } } catch ( Exception e1 ) { String msg = statusMessagePrefix ( ) + "ERROR: unable to create output instances structure." ; if ( m_log != null ) { m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + e1 . getMessage ( ) ) ; } stop ( ) ; e1 . printStackTrace ( ) ; m_busy = false ; return ; } } m_busy = false ; return ; } if ( e . getStatus ( ) == InstanceEvent . BATCH_FINISHED || e . getStatus ( ) == InstanceEvent . INSTANCE_AVAILABLE ) { Instance currentI = e . getInstance ( ) ; if ( m_completeHeader == null ) { if ( currentI != null ) { ArffSaver saver = m_incrementalSavers . get ( e . getSource ( ) ) ; if ( saver == null ) { saver = new ArffSaver ( ) ; try { File tmpFile = File . createTempFile ( "weka" , ".arff" ) ; saver . setFile ( tmpFile ) ; saver . setRetrieval ( weka . core . converters . Saver . INCREMENTAL ) ; saver . setInstances ( new Instances ( currentI . dataset ( ) , 0 ) ) ; m_incrementalSavers . put ( e . getSource ( ) , saver ) ; } catch ( IOException e1 ) { stop ( ) ; e1 . printStackTrace ( ) ; String msg = statusMessagePrefix ( ) + "ERROR: unable to save instance to temp file" ; if ( m_log != null ) { m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + e1 . getMessage ( ) ) ; } m_busy = false ; return ; } } try { saver . writeIncremental ( currentI ) ; if ( e . getStatus ( ) == InstanceEvent . BATCH_FINISHED ) { m_finishedCount ++ ; } } catch ( IOException e1 ) { stop ( ) ; e1 . printStackTrace ( ) ; String msg = statusMessagePrefix ( ) + "ERROR: unable to save instance to temp file" ; if ( m_log != null ) { m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + e1 . getMessage ( ) ) ; } m_busy = false ; return ; } } } else { if ( currentI != null ) { int code = InstanceEvent . INSTANCE_AVAILABLE ; if ( e . getStatus ( ) == InstanceEvent . BATCH_FINISHED ) { m_finishedCount ++ ; if ( m_finishedCount == m_listenees . size ( ) ) { code = InstanceEvent . BATCH_FINISHED ; } } Instance newI = makeOutputInstance ( m_completeHeader , currentI ) ; m_ie . setStatus ( code ) ; m_ie . setInstance ( newI ) ; notifyInstanceListeners ( m_ie ) ; m_incrementalCounter ++ ; if ( m_incrementalCounter % 10000 == 0 ) { if ( m_log != null ) { m_log . statusMessage ( statusMessagePrefix ( ) + "Processed " + m_incrementalCounter + " instances" ) ; } } if ( code == InstanceEvent . BATCH_FINISHED ) { if ( m_log != null ) { m_log . statusMessage ( statusMessagePrefix ( ) + "Finished" ) ; } m_completed = null ; m_incrementalSavers = null ; m_incrementalCounter = 0 ; m_completeHeader = null ; m_finishedCount = 0 ; } } } } m_busy = false ; }
accept and process an instance event
