IoT CAS Frameworks/IOT_Server/Multi-point_Event_Router_Framework/Event_Receiver_Mapper/EventReceiverMapper/src/com/prokosha/eventreceivermapper/EventProcessor.java
public synchronized void sendEvent(String eventType, String event) {
        String customerId = getCustomerID(event);
        if (gszEventProcessorName.equalsIgnoreCase("etl")) {
            try {
                //System.out.println("Sending event to ETL through method invocation");
                String status = szETLAdapter.dumpCEPEvent(event);
                String splitst[] = status.split(":");
                if ((status.equalsIgnoreCase("null:0") || (status.equalsIgnoreCase("0")))) {
                    System.out.println("Error in ETL insert metric, logging eventdata to reprocess later");
                    log.error("Error in ETL insert metric, logging eventdata to reprocess later");
                    evtLogger.info(event);
                }else if (splitst.length > 1) {
                    if (Integer.parseInt(splitst[1]) == -1) {
                        System.out.println("Error in ETL insert metric, logging eventdata to reprocess later");
                        log.error("Error in ETL insert metric, logging eventdata to reprocess later");
                        evtLogger.info(event);
                    }
                }
                System.out.println("ETL return status==>>" + status);
            } catch (Exception e) {
                System.out.println("Error in sending event to ETL through method invocation");
                e.printStackTrace();
            }
            return;
        }
        //System.out.println("gszEventProcessorName={" + gszEventProcessorName + "} and customerId{" + customerId + "} evaluating expression  == "  + "gszEventProcessorName.contains(['customerId']) > " + gszEventProcessorName.contains("[" + customerId + "]"));
        if (EventReceiverMapper.isForMultipleClient() && !gszEventProcessorName.contains("[" + customerId + "]")) {
            int unknowncustomerID = 0;
            int totalCustomers = EventReceiverMapper.gszArrProcessorGroup.length;
            for (String str : EventReceiverMapper.gszArrProcessorGroup) {
                if (!str.contains("[" + customerId + "]")) {
                    unknowncustomerID++;
                }
            }
            if (unknowncustomerID >= totalCustomers) {
                try {
                    throw new Exception("UnknownCustomerIDException: Unknown CustomerID '" + customerId + "' in Event-> " + event + " is found.");
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            return;
        }
        //System.out.println("for eventType : " + eventType + " sender of name : " + gszEventProcessorName + " sending data : " + event);
        //System.out.println("sending data to " + gszEventProcessorHost + "@" + gszEventProcessorPort + "....");
        try {
            if (writer == null) {
                System.out.println("writer Object is null; could send data to " + gszEventProcessorHost + "@" + gszEventProcessorPort + "....");
                closeDataProcessor();
                init();
            }
            if (writer != null) {
                if (writer.checkError()) {
                    closeDataProcessor();
                    init();
                }
            }
            if (!skt.getChannel().isOpen()) {
                closeDataProcessor();
                init();
            }
            if ((!writer.checkError()) && entered) {
                System.out.println("Analysing status for " + gszEventProcessorHost + "@" + gszEventProcessorPort);
                System.out.println("\tskt.isClosed()         : " + skt.isClosed());
                System.out.println("\tskt.isConnected()      : " + skt.isConnected());
                System.out.println("\tskt.isOutputShutdown() : " + skt.isOutputShutdown());
                //writer.notify();
                //writer.flush();
                System.out.println("soket is reachable : " + skt.getInetAddress().isReachable(5000));
                System.out.println("writer.checkError() = " + writer.checkError());
                //skt.
                writer.println(event);
                /*if (!dataStorage.isEmpty()) {
                 System.out.println("sending " + dataStorage.size() + " data (previous data) to " + gszEventProcessorHost + "@" + iport);
                 for (String dataSt : dataStorage) {
                 writer.flush();
                 writer.println(dataSt);
                 }
                 }*/
                writer.flush();
                event = null;
            } else {
                System.out.println("Event not sent to " + gszEventProcessorHost + "@" + iport);
                //dataStorage.add(event);
                entered = false;
                init();
            }
            System.out.println("openPortCount = " + openPortCount);
            //closeDataSender();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
/**
     * sends event data to the host mapped by the current EventProcessor
     *
     * @param eventType Type of event
     * @param event String data of event
     */
sends event data to the host mapped by the current EventProcessor
@param eventType Type of event
@param event String data of event
sends event data to the host mapped by the current EventProcessor @param eventType Type of event @param event String data of event
public synchronized void sendEvent(String eventType, String event) {
        String customerId = getCustomerID(event);
        if (gszEventProcessorName.equalsIgnoreCase("etl")) {
            try {
                String status = szETLAdapter.dumpCEPEvent(event);
                String splitst[] = status.split(":");
                if ((status.equalsIgnoreCase("null:0") || (status.equalsIgnoreCase("0")))) {
                    System.out.println("Error in ETL insert metric, logging eventdata to reprocess later");
                    log.error("Error in ETL insert metric, logging eventdata to reprocess later");
                    evtLogger.info(event);
                }else if (splitst.length > 1) {
                    if (Integer.parseInt(splitst[1]) == -1) {
                        System.out.println("Error in ETL insert metric, logging eventdata to reprocess later");
                        log.error("Error in ETL insert metric, logging eventdata to reprocess later");
                        evtLogger.info(event);
                    }
                }
                System.out.println("ETL return status==>>" + status);
            } catch (Exception e) {
                System.out.println("Error in sending event to ETL through method invocation");
                e.printStackTrace();
            }
            return;
        }
        if (EventReceiverMapper.isForMultipleClient() && !gszEventProcessorName.contains("[" + customerId + "]")) {
            int unknowncustomerID = 0;
            int totalCustomers = EventReceiverMapper.gszArrProcessorGroup.length;
            for (String str : EventReceiverMapper.gszArrProcessorGroup) {
                if (!str.contains("[" + customerId + "]")) {
                    unknowncustomerID++;
                }
            }
            if (unknowncustomerID >= totalCustomers) {
                try {
                    throw new Exception("UnknownCustomerIDException: Unknown CustomerID '" + customerId + "' in Event-> " + event + " is found.");
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            return;
        }
        try {
            if (writer == null) {
                System.out.println("writer Object is null; could send data to " + gszEventProcessorHost + "@" + gszEventProcessorPort + "....");
                closeDataProcessor();
                init();
            }
            if (writer != null) {
                if (writer.checkError()) {
                    closeDataProcessor();
                    init();
                }
            }
            if (!skt.getChannel().isOpen()) {
                closeDataProcessor();
                init();
            }
            if ((!writer.checkError()) && entered) {
                System.out.println("Analysing status for " + gszEventProcessorHost + "@" + gszEventProcessorPort);
                System.out.println("\tskt.isClosed()         : " + skt.isClosed());
                System.out.println("\tskt.isConnected()      : " + skt.isConnected());
                System.out.println("\tskt.isOutputShutdown() : " + skt.isOutputShutdown());
                System.out.println("soket is reachable : " + skt.getInetAddress().isReachable(5000));
                System.out.println("writer.checkError() = " + writer.checkError());
                writer.println(event);
                /*if (!dataStorage.isEmpty()) {
                 System.out.println("sending " + dataStorage.size() + " data (previous data) to " + gszEventProcessorHost + "@" + iport);
                 for (String dataSt : dataStorage) {
                 writer.flush();
                 writer.println(dataSt);
                 }
                 }*/
                writer.flush();
                event = null;
            } else {
                System.out.println("Event not sent to " + gszEventProcessorHost + "@" + iport);
                entered = false;
                init();
            }
            System.out.println("openPortCount = " + openPortCount);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
public synchronized void sendEvent ( String eventType , String event ) { String customerId = getCustomerID ( event ) ; if ( gszEventProcessorName . equalsIgnoreCase ( "etl" ) ) { try { String status = szETLAdapter . dumpCEPEvent ( event ) ; String splitst [ ] = status . split ( ":" ) ; if ( ( status . equalsIgnoreCase ( "null:0" ) || ( status . equalsIgnoreCase ( "0" ) ) ) ) { System . out . println ( "Error in ETL insert metric, logging eventdata to reprocess later" ) ; log . error ( "Error in ETL insert metric, logging eventdata to reprocess later" ) ; evtLogger . info ( event ) ; } else if ( splitst . length > 1 ) { if ( Integer . parseInt ( splitst [ 1 ] ) == - 1 ) { System . out . println ( "Error in ETL insert metric, logging eventdata to reprocess later" ) ; log . error ( "Error in ETL insert metric, logging eventdata to reprocess later" ) ; evtLogger . info ( event ) ; } } System . out . println ( "ETL return status==>>" + status ) ; } catch ( Exception e ) { System . out . println ( "Error in sending event to ETL through method invocation" ) ; e . printStackTrace ( ) ; } return ; } if ( EventReceiverMapper . isForMultipleClient ( ) && ! gszEventProcessorName . contains ( "[" + customerId + "]" ) ) { int unknowncustomerID = 0 ; int totalCustomers = EventReceiverMapper . gszArrProcessorGroup . length ; for ( String str : EventReceiverMapper . gszArrProcessorGroup ) { if ( ! str . contains ( "[" + customerId + "]" ) ) { unknowncustomerID ++ ; } } if ( unknowncustomerID >= totalCustomers ) { try { throw new Exception ( "UnknownCustomerIDException: Unknown CustomerID '" + customerId + "' in Event-> " + event + " is found." ) ; } catch ( Exception ex ) { ex . printStackTrace ( ) ; } } return ; } try { if ( writer == null ) { System . out . println ( "writer Object is null; could send data to " + gszEventProcessorHost + "@" + gszEventProcessorPort + "...." ) ; closeDataProcessor ( ) ; init ( ) ; } if ( writer != null ) { if ( writer . checkError ( ) ) { closeDataProcessor ( ) ; init ( ) ; } } if ( ! skt . getChannel ( ) . isOpen ( ) ) { closeDataProcessor ( ) ; init ( ) ; } if ( ( ! writer . checkError ( ) ) && entered ) { System . out . println ( "Analysing status for " + gszEventProcessorHost + "@" + gszEventProcessorPort ) ; System . out . println ( "\tskt.isClosed()         : " + skt . isClosed ( ) ) ; System . out . println ( "\tskt.isConnected()      : " + skt . isConnected ( ) ) ; System . out . println ( "\tskt.isOutputShutdown() : " + skt . isOutputShutdown ( ) ) ; System . out . println ( "soket is reachable : " + skt . getInetAddress ( ) . isReachable ( 5000 ) ) ; System . out . println ( "writer.checkError() = " + writer . checkError ( ) ) ; writer . println ( event ) ; /*if (!dataStorage.isEmpty()) {
                 System.out.println("sending " + dataStorage.size() + " data (previous data) to " + gszEventProcessorHost + "@" + iport);
                 for (String dataSt : dataStorage) {
                 writer.flush();
                 writer.println(dataSt);
                 }
                 }*/ writer . flush ( ) ; event = null ; } else { System . out . println ( "Event not sent to " + gszEventProcessorHost + "@" + iport ) ; entered = false ; init ( ) ; } System . out . println ( "openPortCount = " + openPortCount ) ; } catch ( Exception ex ) { ex . printStackTrace ( ) ; } }
sends event data to the host mapped by the current EventProcessor
 @param eventType Type of event
 @param event String data of event
sends event data to the host mapped by the current EventProcessor @param eventType Type of event @param event String data of event
//System.out.println("Sending event to ETL through method invocation"); //System.out.println("gszEventProcessorName={" + gszEventProcessorName + "} and customerId{" + customerId + "} evaluating expression  == "  + "gszEventProcessorName.contains(['customerId']) > " + gszEventProcessorName.contains("[" + customerId + "]")); //System.out.println("for eventType : " + eventType + " sender of name : " + gszEventProcessorName + " sending data : " + event); //System.out.println("sending data to " + gszEventProcessorHost + "@" + gszEventProcessorPort + "...."); //writer.notify(); //writer.flush(); //skt. //dataStorage.add(event); //closeDataSender();