@SuppressWarnings("unused")
    private String getEventName(byte[] inputs) {
        String  s = new String(inputs);
        //  Remove Tango host
        int pos = s.lastIndexOf('.');
        for (int i=0 ; i<4 ; i++)
            pos = s.lastIndexOf('/', pos-1);
        return s.substring(pos+1);
    }
    //===============================================================
    //===============================================================
    private void checkEventMessage(byte[][] inputs) throws Exception {

        if (inputs.length<NbFields) {   // || inputs[EndianIdx].length==0) {
            System.err.println("NbFields=" + NbFields);
            eventDrift = 4 - inputs.length;
            Except.throw_exception("Api_BadParameterException",
                    "Cannot decode event  (message size !)",
                    "ZmqMainThread.checkEventMessage()");
        }
        if (new String(inputs[NameIdx]).startsWith("tango://")) {
            //if (new String(inputs[NameIdx]).toLowerCase().contains("position"))
            //    System.out.println("checkEventMessage(): Receive " +new String(inputs[NameIdx]));
            return; //  OK
        }

        //  Check what is first input ?
        byte[]  bytes = inputs[NameIdx];
        if (bytes.length==1) {
            //  Endianess
            eventDrift = 3;
            Except.throw_exception("Api_BadParameterException",
                    "Cannot decode event  (start with endianess)",
                    "ZmqMainThread.checkEventMessage()");
        }
        else
        if (bytes[0]==0xc && bytes[1]==0x0 && bytes[2]==0xd && bytes[3]==0xe) {
            //  DATA
            eventDrift = 1;
            Except.throw_exception("Api_BadParameterException",
                    "Cannot decode event  (start with data)",
                    "ZmqMainThread.checkEventMessage()");
        }
        else {
            //  Specifications
            eventDrift = 2;
            Except.throw_exception("Api_BadParameterException",
                    "Cannot decode event  (start with specifications)",
                    "ZmqMainThread.checkEventMessage()");
        }
    }
    //===============================================================
    /**
     * Manage events. Extract data and push to callback
     * @param inputs received messages
     * @throws DevFailed if cannot extract event type or data from messages.
     */
    //===============================================================
    private void manageEvent(byte[][] inputs) throws DevFailed {

        try {
            //  Check if inputs are coherent
            checkEventMessage(inputs);

            //  Decode receive data parts
            //String  eventName = getEventName(inputs[NameIdx]);
            String  eventName = new String(inputs[NameIdx]);
            boolean littleEndian = true;
            if (inputs[EndianIdx].length>0) {
                //  Sometimes inputs[EndianIdx] could be empty (fixed in c++ 8.1)
                littleEndian = (inputs[EndianIdx][0]!=0);
            }
            ZmqCallInfo zmqCallInfo =
                    ZMQutils.deMarshallZmqCallInfo(inputs[ZmqInfoIdx], littleEndian);
            if (zmqCallInfo!=null) {
                manageEventValue(eventName, ApiUtil.toLongUnsigned(zmqCallInfo.ctr),
                        inputs[ValueIdx], littleEndian, zmqCallInfo.call_is_except);
            }
            else
                throw new Exception("DeMarshalling returns null");
        }
        catch (Exception e) {
            if (e instanceof DevFailed)
                throw (DevFailed) e;
            e.printStackTrace();
            Except.throw_exception("Api_CatchException",
                    "API catch a " + e.toString() + " exception",
                    "ZmqMainThread.manageEvent()");
        }
    }
    //===============================================================
    //===============================================================
    private EventCallBackStruct getEventCallBackStruct(String eventName) {
        List<String> possibleTangoHosts = EventConsumer.possibleTangoHosts;
        Hashtable<String, EventCallBackStruct> callbackMap = ZmqEventConsumer.getEventCallbackMap();
        if (callbackMap.containsKey(eventName)) {
            return callbackMap.get(eventName);
        }
        //  Check with other TangoHosts using possibleTangoHosts as header
        int index = eventName.indexOf("//");
        if (index>0) {
            index = eventName.indexOf('/', index+2); //  "//".length()
            for (String possibleTangoHost : possibleTangoHosts) {
                String key = possibleTangoHost + eventName.substring(index);
                if (callbackMap.containsKey(key)) {
                    return callbackMap.get(key);
                }
            }
        }

        //  Not found
        /*  Display table content
		if (eventName.contains("maxtempchange")) {
			System.err.println("===============================================================");
			System.err.println(eventName + " NOT FOUND");
			System.err.println("keys are:");
        	Enumeration<String> keys = callbackMap.keys();
        	while (keys.hasMoreElements()) {
            	System.err.println(keys.nextElement());
        	}
		}
        */
        return null;
    }
    //===============================================================
    //===============================================================
    private void manageEventValue(String eventName,
                                  long eventCounter,
                                  byte[] recData,
                                  boolean littleEndian,
                                  boolean isExcept) throws  DevFailed {
        //System.out.println("Event name  = " + eventName);
        EventCallBackStruct callBackStruct = getEventCallBackStruct(eventName);
        if (callBackStruct!=null) {
            DeviceAttribute attributeValue  = null;
            DevicePipe      devicePipe      = null;
            AttributeInfoEx attributeConfig = null;
            AttDataReady    dataReady       = null;
            DeviceInterface deviceInterface = null;
            DevError[]      devErrorList    = null;

            //  Manage ZMQ counter (queue has reached HWM ?)
            boolean pushTheEvent =
                    manageEventCounter(callBackStruct, eventName, eventCounter);
            ZMQutils.zmqEventTrace("ZMQ event from " + eventName);

            //  Check if Value part is a DevFailed
            if (isExcept) {
                devErrorList = ZMQutils.deMarshallErrorList(recData, littleEndian);
            }
            else {
                Hashtable<String, EventChannelStruct> channelMap = EventConsumer.getChannelMap();
                EventChannelStruct eventChannelStruct = channelMap.get(callBackStruct.channel_name);
                if (eventChannelStruct!=null) {
                    try {
                        //  Needs idl version to de marshall
                        int idl = callBackStruct.device.get_idl_version();

                        //  Else check event type
                        switch (ZMQutils.getEventType(eventName)) {
                            case TangoConst.ATT_CONF_EVENT:
                                attributeConfig =
                                        ZMQutils.deMarshallAttributeConfig(recData, littleEndian, idl);
                                break;
                            case TangoConst.PIPE_EVENT:
                                devicePipe = ZMQutils.deMarshallPipe(recData, littleEndian, idl);
                                break;
                            case TangoConst.DATA_READY_EVENT:
                                dataReady = ZMQutils.deMarshallAttDataReady(recData, littleEndian);
                                break;
                            case TangoConst.INTERFACE_CHANGE:
                                deviceInterface = ZMQutils.deMarshallAttInterfaceChange(recData, littleEndian);
                                break;
                            default:
                                attributeValue = ZMQutils.deMarshallAttribute(recData, littleEndian, idl);
                         }
                    }
                    catch(DevFailed e) {
                        //  convert de marshall
                        devErrorList = e.errors;
                    }
                }
            }
            if (pushTheEvent) {
                //	Build and push event data
                String  deviceName = getDeviceName(eventName);
                pushEventData(callBackStruct,
                        new EventData(callBackStruct.device,
                                deviceName, eventName,
                                callBackStruct.event_type, EventData.ZMQ_EVENT,
                                attributeValue, devicePipe,
                                attributeConfig, dataReady,
                                deviceInterface, devErrorList));
            }
        }
        else
            System.err.println(eventName + " ?  NOT FOUND");
    }
    //===============================================================
    /**
     * Manage the event counter
     * @param callBackStruct    the event callback structure
     * @param eventName         the event name
     * @param eventCounter    the event counter to manage
     * @return true if the event must be pushed.
     * @throws DevFailed if at least one event has been lost
     */
    //===============================================================
    private boolean manageEventCounter(EventCallBackStruct callBackStruct,
                                    String eventName,
                                    long eventCounter) throws DevFailed {
        long    previousCounter = callBackStruct.getZmqCounter();
        //  Is it the first call ?
        if (previousCounter==Long.MAX_VALUE) {
            callBackStruct.setZmqCounter(eventCounter);
            //  There is NO synchronous call for DataReady event !!!!
            if (callBackStruct.event_name.equals(
                    TangoConst.eventNames[TangoConst.DATA_READY_EVENT])) {
                callBackStruct.setSynchronousDone(true);
            }
            //  To be sure to have first event after synchronous call,
            //      wait the event pushed in dedicated thread.
            int timeout = 5000;
            for (int i=0 ; !callBackStruct.isSynchronousDone() && i<timeout ; i++) {
                try { Thread.sleep(1); } catch (InterruptedException e) { /* */ }
            }
            return true;
        }

        long    delta = eventCounter - previousCounter;
        //  If delta==0 --> already receive (ZMQ bug)
        if (delta==0) {
            callBackStruct.setZmqCounter(eventCounter);
            return false;
        }
        //  eventCounter<=0  -> reconnection
        if (eventCounter<=0) {
            callBackStruct.setZmqCounter(eventCounter);
            return false;
        }

        //  If delta==1 -> It is OK nothing lost
        if (delta==1) {
            callBackStruct.setZmqCounter(eventCounter);
            return true;
        }

        //  if delta<0  --> integer overflow
        if (delta<0) {
            long    maxCounter = ApiUtil.toLongUnsigned(-1);
            long    delta2 = maxCounter - delta;
            //  If delta2==1 -> It is OK nothing lost
            if (delta2==1) {
                callBackStruct.setZmqCounter(eventCounter);
                return true;
            }
        }
        //  Else
        //  At least one event has been lost, push a DevError event
        long nb = eventCounter - (previousCounter+1);
        DevError[]      devErrorList    = new DevError[] {
                    new DevError("Api_MissedEvents", ErrSeverity.ERR,
                        "Missed " + nb + " events ("  +
                                eventCounter + "-" + (previousCounter+1) +
                                ") ! ZMQ queue has reached HWM or resynchronize ?",
                        "ZmqMainThread.manageEventCounter()") };

        //	Build and push event data
        String  deviceName = getDeviceName(eventName);
        pushEventData(callBackStruct,
                new EventData(callBackStruct.device,
                        deviceName, eventName,
                        callBackStruct.event_type, EventData.ZMQ_EVENT,
                        null, null, null, null, null, devErrorList));
        callBackStruct.setZmqCounter(eventCounter);
        return true;
    }
    //===============================================================
    //===============================================================
    private void pushEventData(EventCallBackStruct callBackStruct, EventData eventData) {

        if (callBackStruct.use_ev_queue) {
            EventQueue ev_queue = callBackStruct.device.getEventQueue();
            ev_queue.insert_event(eventData);
        } else if (callBackStruct.callback != null) {
            callBackStruct.callback.push_event(eventData);
        }
    }
    //===============================================================
    /**
     * Manage heartbeat
     * @param inputs received messages
     * @throws DevFailed if cannot get ZmqEventConsumer instance
     */
    //===============================================================
    private void manageHeartbeat(byte[][] inputs) throws DevFailed{
        //  First part is heartbeat name
        String  name = new String(inputs[NameIdx]);
        //System.out.println("heartbeat : " + name);

        //  Check if name is coherent
        int start = name.indexOf("dserver/");
        if (start<0) {
            //  ToDo
            long    t = System.currentTimeMillis();
            System.err.println(formatTime(t) + ":\n heartbeat: " + name +
                    " cannot be parsed ! length=" + inputs[NameIdx].length);
            ZMQutils.dump(inputs[NameIdx]);

            //  Check if endianess or specif
            if (inputs[NameIdx].length==1)  //  Endianess
                heartbeatDrift = 2;
            else
                heartbeatDrift = 1;
            return;
        }

        //  Get only device name (without event type.
        int end   = name.lastIndexOf('.');
        name = name.substring(0, end);
        //System.out.println("-----------> Receive Heartbeat for " + name);
        ZmqEventConsumer.getInstance().push_structured_event_heartbeat(name);

        //  Second one is endianess
        if (inputs[EndianIdx].length==0) {
            System.err.println("heartbeat "+ name + ":   endianess is missing !!!");
        }
        /*
        else {
             NOT USED
        }
        */
    }
@ SuppressWarnings ( "unused" ) private String getEventName ( byte [ ] inputs ) { String s = new String ( inputs ) ; int pos = s . lastIndexOf ( '.' ) ; for ( int i = 0 ; i < 4 ; i ++ ) pos = s . lastIndexOf ( '/' , pos - 1 ) ; return s . substring ( pos + 1 ) ; } private void checkEventMessage ( byte [ ] [ ] inputs ) throws Exception { if ( inputs . length < NbFields ) { System . err . println ( "NbFields=" + NbFields ) ; eventDrift = 4 - inputs . length ; Except . throw_exception ( "Api_BadParameterException" , "Cannot decode event  (message size !)" , "ZmqMainThread.checkEventMessage()" ) ; } if ( new String ( inputs [ NameIdx ] ) . startsWith ( "tango://" ) ) { return ; } byte [ ] bytes = inputs [ NameIdx ] ; if ( bytes . length == 1 ) { eventDrift = 3 ; Except . throw_exception ( "Api_BadParameterException" , "Cannot decode event  (start with endianess)" , "ZmqMainThread.checkEventMessage()" ) ; } else if ( bytes [ 0 ] == 0xc && bytes [ 1 ] == 0x0 && bytes [ 2 ] == 0xd && bytes [ 3 ] == 0xe ) { eventDrift = 1 ; Except . throw_exception ( "Api_BadParameterException" , "Cannot decode event  (start with data)" , "ZmqMainThread.checkEventMessage()" ) ; } else { eventDrift = 2 ; Except . throw_exception ( "Api_BadParameterException" , "Cannot decode event  (start with specifications)" , "ZmqMainThread.checkEventMessage()" ) ; } } private void manageEvent ( byte [ ] [ ] inputs ) throws DevFailed { try { checkEventMessage ( inputs ) ; String eventName = new String ( inputs [ NameIdx ] ) ; boolean littleEndian = true ; if ( inputs [ EndianIdx ] . length > 0 ) { littleEndian = ( inputs [ EndianIdx ] [ 0 ] != 0 ) ; } ZmqCallInfo zmqCallInfo = ZMQutils . deMarshallZmqCallInfo ( inputs [ ZmqInfoIdx ] , littleEndian ) ; if ( zmqCallInfo != null ) { manageEventValue ( eventName , ApiUtil . toLongUnsigned ( zmqCallInfo . ctr ) , inputs [ ValueIdx ] , littleEndian , zmqCallInfo . call_is_except ) ; } else throw new Exception ( "DeMarshalling returns null" ) ; } catch ( Exception e ) { if ( e instanceof DevFailed ) throw ( DevFailed ) e ; e . printStackTrace ( ) ; Except . throw_exception ( "Api_CatchException" , "API catch a " + e . toString ( ) + " exception" , "ZmqMainThread.manageEvent()" ) ; } } private EventCallBackStruct getEventCallBackStruct ( String eventName ) { List < String > possibleTangoHosts = EventConsumer . possibleTangoHosts ; Hashtable < String , EventCallBackStruct > callbackMap = ZmqEventConsumer . getEventCallbackMap ( ) ; if ( callbackMap . containsKey ( eventName ) ) { return callbackMap . get ( eventName ) ; } int index = eventName . indexOf ( "//" ) ; if ( index > 0 ) { index = eventName . indexOf ( '/' , index + 2 ) ; for ( String possibleTangoHost : possibleTangoHosts ) { String key = possibleTangoHost + eventName . substring ( index ) ; if ( callbackMap . containsKey ( key ) ) { return callbackMap . get ( key ) ; } } } return null ; } private void manageEventValue ( String eventName , long eventCounter , byte [ ] recData , boolean littleEndian , boolean isExcept ) throws DevFailed { EventCallBackStruct callBackStruct = getEventCallBackStruct ( eventName ) ; if ( callBackStruct != null ) { DeviceAttribute attributeValue = null ; DevicePipe devicePipe = null ; AttributeInfoEx attributeConfig = null ; AttDataReady dataReady = null ; DeviceInterface deviceInterface = null ; DevError [ ] devErrorList = null ; boolean pushTheEvent = manageEventCounter ( callBackStruct , eventName , eventCounter ) ; ZMQutils . zmqEventTrace ( "ZMQ event from " + eventName ) ; if ( isExcept ) { devErrorList = ZMQutils . deMarshallErrorList ( recData , littleEndian ) ; } else { Hashtable < String , EventChannelStruct > channelMap = EventConsumer . getChannelMap ( ) ; EventChannelStruct eventChannelStruct = channelMap . get ( callBackStruct . channel_name ) ; if ( eventChannelStruct != null ) { try { int idl = callBackStruct . device . get_idl_version ( ) ; switch ( ZMQutils . getEventType ( eventName ) ) { case TangoConst . ATT_CONF_EVENT : attributeConfig = ZMQutils . deMarshallAttributeConfig ( recData , littleEndian , idl ) ; break ; case TangoConst . PIPE_EVENT : devicePipe = ZMQutils . deMarshallPipe ( recData , littleEndian , idl ) ; break ; case TangoConst . DATA_READY_EVENT : dataReady = ZMQutils . deMarshallAttDataReady ( recData , littleEndian ) ; break ; case TangoConst . INTERFACE_CHANGE : deviceInterface = ZMQutils . deMarshallAttInterfaceChange ( recData , littleEndian ) ; break ; default : attributeValue = ZMQutils . deMarshallAttribute ( recData , littleEndian , idl ) ; } } catch ( DevFailed e ) { devErrorList = e . errors ; } } } if ( pushTheEvent ) { String deviceName = getDeviceName ( eventName ) ; pushEventData ( callBackStruct , new EventData ( callBackStruct . device , deviceName , eventName , callBackStruct . event_type , EventData . ZMQ_EVENT , attributeValue , devicePipe , attributeConfig , dataReady , deviceInterface , devErrorList ) ) ; } } else System . err . println ( eventName + " ?  NOT FOUND" ) ; } private boolean manageEventCounter ( EventCallBackStruct callBackStruct , String eventName , long eventCounter ) throws DevFailed { long previousCounter = callBackStruct . getZmqCounter ( ) ; if ( previousCounter == Long . MAX_VALUE ) { callBackStruct . setZmqCounter ( eventCounter ) ; if ( callBackStruct . event_name . equals ( TangoConst . eventNames [ TangoConst . DATA_READY_EVENT ] ) ) { callBackStruct . setSynchronousDone ( true ) ; } int timeout = 5000 ; for ( int i = 0 ; ! callBackStruct . isSynchronousDone ( ) && i < timeout ; i ++ ) { try { Thread . sleep ( 1 ) ; } catch ( InterruptedException e ) { } } return true ; } long delta = eventCounter - previousCounter ; if ( delta == 0 ) { callBackStruct . setZmqCounter ( eventCounter ) ; return false ; } if ( eventCounter <= 0 ) { callBackStruct . setZmqCounter ( eventCounter ) ; return false ; } if ( delta == 1 ) { callBackStruct . setZmqCounter ( eventCounter ) ; return true ; } if ( delta < 0 ) { long maxCounter = ApiUtil . toLongUnsigned ( - 1 ) ; long delta2 = maxCounter - delta ; if ( delta2 == 1 ) { callBackStruct . setZmqCounter ( eventCounter ) ; return true ; } } long nb = eventCounter - ( previousCounter + 1 ) ; DevError [ ] devErrorList = new DevError [ ] { new DevError ( "Api_MissedEvents" , ErrSeverity . ERR , "Missed " + nb + " events (" + eventCounter + "-" + ( previousCounter + 1 ) + ") ! ZMQ queue has reached HWM or resynchronize ?" , "ZmqMainThread.manageEventCounter()" ) } ; String deviceName = getDeviceName ( eventName ) ; pushEventData ( callBackStruct , new EventData ( callBackStruct . device , deviceName , eventName , callBackStruct . event_type , EventData . ZMQ_EVENT , null , null , null , null , null , devErrorList ) ) ; callBackStruct . setZmqCounter ( eventCounter ) ; return true ; } private void pushEventData ( EventCallBackStruct callBackStruct , EventData eventData ) { if ( callBackStruct . use_ev_queue ) { EventQueue ev_queue = callBackStruct . device . getEventQueue ( ) ; ev_queue . insert_event ( eventData ) ; } else if ( callBackStruct . callback != null ) { callBackStruct . callback . push_event ( eventData ) ; } } private void manageHeartbeat ( byte [ ] [ ] inputs ) throws DevFailed { String name = new String ( inputs [ NameIdx ] ) ; int start = name . indexOf ( "dserver/" ) ; if ( start < 0 ) { long t = System . currentTimeMillis ( ) ; System . err . println ( formatTime ( t ) + ":\n heartbeat: " + name + " cannot be parsed ! length=" + inputs [ NameIdx ] . length ) ; ZMQutils . dump ( inputs [ NameIdx ] ) ; if ( inputs [ NameIdx ] . length == 1 ) heartbeatDrift = 2 ; else heartbeatDrift = 1 ; return ; } int end = name . lastIndexOf ( '.' ) ; name = name . substring ( 0 , end ) ; ZmqEventConsumer . getInstance ( ) . push_structured_event_heartbeat ( name ) ; if ( inputs [ EndianIdx ] . length == 0 ) { System . err . println ( "heartbeat " + name + ":   endianess is missing !!!" ) ; } }
===============================================================