@Override public synchronized void acceptDataSet(DataSetEvent e){
  m_busy=true;
  if (m_completed == null) {
    m_completed=new HashMap<Object,Instances>();
    m_tempBatchFiles=new HashMap<Object,File>();
  }
  Object source=e.getSource();
  if (m_completed.containsKey(source)) {
    if (m_log != null && !e.isStructureOnly()) {
      String msg=statusMessagePrefix() + "Resetting appender.";
      m_log.statusMessage(msg);
      m_log.logMessage("[Appender] " + msg + " New batch for an incoming connection "+ "detected before "+ "all incoming connections have sent data!");
    }
    m_completed=new HashMap<Object,Instances>();
    m_tempBatchFiles=new HashMap<Object,File>();
  }
  Instances header=new Instances(e.getDataSet(),0);
  m_completed.put(source,header);
  try {
    File tmpF=File.createTempFile("weka",SerializedInstancesLoader.FILE_EXTENSION);
    tmpF.deleteOnExit();
    ObjectOutputStream oos=new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream(tmpF)));
    oos.writeObject(e.getDataSet());
    oos.flush();
    oos.close();
    m_tempBatchFiles.put(source,tmpF);
  }
 catch (  IOException e1) {
    stop();
    e1.printStackTrace();
    String msg=statusMessagePrefix() + "ERROR: unable to save batch instances to temp file";
    if (m_log != null) {
      m_log.statusMessage(msg);
      m_log.logMessage("[Appender] " + e1.getMessage());
    }
    m_busy=false;
    return;
  }
  if (m_completed.size() == m_listenees.size()) {
    try {
      Instances output=makeOutputHeader();
      if (m_log != null) {
        String msg=statusMessagePrefix() + "Making output header";
        m_log.statusMessage(msg);
        m_log.logMessage("[Appender] " + msg);
      }
      for (      File f : m_tempBatchFiles.values()) {
        ObjectInputStream ois=new ObjectInputStream(new BufferedInputStream(new FileInputStream(f)));
        Instances temp=(Instances)ois.readObject();
        ois.close();
        for (int i=0; i < temp.numInstances(); i++) {
          Instance converted=makeOutputInstance(output,temp.instance(i));
          output.add(converted);
        }
      }
      DataSetEvent d=new DataSetEvent(this,output);
      notifyDataListeners(d);
    }
 catch (    Exception ex) {
      stop();
      ex.printStackTrace();
      String msg=statusMessagePrefix() + "ERROR: unable to output appended data set";
      if (m_log != null) {
        m_log.statusMessage(msg);
        m_log.logMessage("[Appender] " + ex.getMessage());
      }
    }
    m_completed=null;
    m_tempBatchFiles=null;
    if (m_log != null) {
      m_log.statusMessage(statusMessagePrefix() + "Finished");
    }
  }
  m_busy=false;
}

@ Override public synchronized void acceptDataSet ( DataSetEvent e ) { m_busy = true ; if ( m_completed == null ) { m_completed = new HashMap < Object , Instances > ( ) ; m_tempBatchFiles = new HashMap < Object , File > ( ) ; } Object source = e . getSource ( ) ; if ( m_completed . containsKey ( source ) ) { if ( m_log != null && ! e . isStructureOnly ( ) ) { String msg = statusMessagePrefix ( ) + "Resetting appender." ; m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + msg + " New batch for an incoming connection " + "detected before " + "all incoming connections have sent data!" ) ; } m_completed = new HashMap < Object , Instances > ( ) ; m_tempBatchFiles = new HashMap < Object , File > ( ) ; } Instances header = new Instances ( e . getDataSet ( ) , 0 ) ; m_completed . put ( source , header ) ; try { File tmpF = File . createTempFile ( "weka" , SerializedInstancesLoader . FILE_EXTENSION ) ; tmpF . deleteOnExit ( ) ; ObjectOutputStream oos = new ObjectOutputStream ( new BufferedOutputStream ( new FileOutputStream ( tmpF ) ) ) ; oos . writeObject ( e . getDataSet ( ) ) ; oos . flush ( ) ; oos . close ( ) ; m_tempBatchFiles . put ( source , tmpF ) ; } catch ( IOException e1 ) { stop ( ) ; e1 . printStackTrace ( ) ; String msg = statusMessagePrefix ( ) + "ERROR: unable to save batch instances to temp file" ; if ( m_log != null ) { m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + e1 . getMessage ( ) ) ; } m_busy = false ; return ; } if ( m_completed . size ( ) == m_listenees . size ( ) ) { try { Instances output = makeOutputHeader ( ) ; if ( m_log != null ) { String msg = statusMessagePrefix ( ) + "Making output header" ; m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + msg ) ; } for ( File f : m_tempBatchFiles . values ( ) ) { ObjectInputStream ois = new ObjectInputStream ( new BufferedInputStream ( new FileInputStream ( f ) ) ) ; Instances temp = ( Instances ) ois . readObject ( ) ; ois . close ( ) ; for ( int i = 0 ; i < temp . numInstances ( ) ; i ++ ) { Instance converted = makeOutputInstance ( output , temp . instance ( i ) ) ; output . add ( converted ) ; } } DataSetEvent d = new DataSetEvent ( this , output ) ; notifyDataListeners ( d ) ; } catch ( Exception ex ) { stop ( ) ; ex . printStackTrace ( ) ; String msg = statusMessagePrefix ( ) + "ERROR: unable to output appended data set" ; if ( m_log != null ) { m_log . statusMessage ( msg ) ; m_log . logMessage ( "[Appender] " + ex . getMessage ( ) ) ; } } m_completed = null ; m_tempBatchFiles = null ; if ( m_log != null ) { m_log . statusMessage ( statusMessagePrefix ( ) + "Finished" ) ; } } m_busy = false ; }
accept and process a data set event
