/* (non-Javadoc)
@see com.ibm.ws.sib.processor.impl.interfaces.ControlHandler#handleControlMessage(com.ibm.ws.sib.trm.topology.Cellule, com.ibm.ws.sib.mfp.control.ControlMessage)

Handle all downstream control messages i.e. target control messages
16672
public void handleControlMessage(SIBUuid8 sourceMEUuid, ControlMessage cMsg)
    throws SIIncorrectCallException, SIErrorException, SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "handleControlMessage", new Object[] { sourceMEUuid, cMsg });

    // Next work out type of ControlMessage and process it
    ControlMessageType type = cMsg.getControlMessageType();

    // First check to see whether this is an "are you flushed" reply.
    // Such messages will not be mappable to a stream ID since we
    // don't yet have a stream data structure (that's why we sent the
    // query in the first place).  Or...these could be stale messages
    // for streams we don't care about.  Either way, handle them
    // elsewhere.
    if(type == ControlMessageType.FLUSHED)
    {
      _targetStreamManager.handleFlushedMessage((ControlFlushed)cMsg);
    }
    else if(type == ControlMessageType.NOTFLUSHED)
    {
      _targetStreamManager.handleNotFlushedMessage((ControlNotFlushed)cMsg);
    }
    else if (type == ControlMessageType.SILENCE)
    {
      _targetStreamManager.handleSilenceMessage((ControlSilence) cMsg);
    }
    else if (type == ControlMessageType.ACKEXPECTED)
    {
      _targetStreamManager.handleAckExpectedMessage((ControlAckExpected) cMsg);
    }
    else
    {
      // Not a recognised type
      // throw exception
    }
  }
1115527
public void handleMessage(
    MessageItem msg,
    TransactionCommon transaction,
    SIBUuid8 sourceMEUuid)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "handleMessage", new Object[]{msg, transaction, sourceMEUuid});
      
    SIErrorException e = new SIErrorException(
      nls.getFormattedMessage(
        "INTERNAL_MESSAGING_ERROR_CWSIP0001",
        new Object[] {
          "com.ibm.ws.sib.processor.impl.DurableInputHandler",
          "1:143:1.52.1.1" },
        null));
    
    // FFDC
    FFDCFilter.processException(
      e,
      "com.ibm.ws.sib.processor.impl.DurableInputHandler.handleMessage",
      "1:150:1.52.1.1",
      this);    
   
    SibTr.exception(tc, e);
    
    SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
      new Object[] {
        "com.ibm.ws.sib.processor.impl.DurableInputHandler",
        "1:158:1.52.1.1" });

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.exit(tc, "handleMessage", e); 
       
    throw e;
  }
28401
@Override
    public void handleMessage(final Message message) throws Fault {
        final InputStream is = message.getContent(InputStream.class);
        if (WireTapHelper.isMessageContentToBeLogged(message, logMessageContent, logMessageContentOverride)) {
            if (null == is) {
                Reader reader = message.getContent(Reader.class);
                if (null != reader) {
                    String encoding = (String) message.get(Message.ENCODING);
                    if (encoding == null) {
                        encoding = "UTF-8";
                    }
                    try {
                        final CachedOutputStream cos = new CachedOutputStream();
                        final Writer writer = new OutputStreamWriter(cos, encoding);
                        IOUtils.copy(reader, writer, 1024);
                        reader.reset();
                        writer.flush();
                        message.setContent(InputStream.class, cos.getInputStream());
                        message.setContent(Reader.class, null);
                        message.setContent(CachedOutputStream.class, cos);
                        closeCachedOutputStream(message, cos);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    } finally {
                        try {
                            reader.reset();
                        } catch (IOException e) {
                        }
                    }
                }
            } else {
                try {
                    final CachedOutputStream cos = new CachedOutputStream();
                    // TODO: We should try to make this streaming
                    //WireTapInputStream wtis = new WireTapInputStream(is, cos);
                    //message.setContent(InputStream.class, wtis);
                    IOUtils.copyAndCloseInput(is, cos);
                    message.setContent(InputStream.class, cos.getInputStream());
                    message.setContent(CachedOutputStream.class, cos);
                    closeCachedOutputStream(message, cos);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }else {
            try {
                final CachedOutputStream cos = new CachedOutputStream();
                cos.write(WireTapHelper.CONTENT_LOGGING_IS_DISABLED.getBytes(Charset.forName("UTF-8")));
                message.setContent(CachedOutputStream.class, cos);
                message.setContent(InputStream.class, is);
                closeCachedOutputStream(message, cos);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }
1258669
@Override
    public void handleMessage(Message message) {

        handlersList = getHanderList(busType, flow);

        if (handlersList == null || handlersList.size() == 0) {

            return;
        }

        ArrayList<Handler> excutedHandlers = new ArrayList<Handler>();
        JAXRS20MessageContextImpl jaxrsMessageContext = new JAXRS20MessageContextImpl(message);
        jaxrsMessageContext.isServerSide = busType.equals(Type.SERVER);
        jaxrsMessageContext.isClientSide = busType.equals(Type.CLIENT);
        jaxrsMessageContext.setProperty(HandlerConstants.FLOW_TYPE, flow.toUpperCase());
        try {
            Iterator<Handler> it = handlersList.iterator();
            while (it.hasNext()) {
                Handler handler = it.next();
                handler.handleMessage(jaxrsMessageContext);
                excutedHandlers.add(handler);
            }
        } catch (Exception e) {
            for (int i = excutedHandlers.size() - 1; i >= 0; i--) {
                excutedHandlers.get(i).handleFault(jaxrsMessageContext);
            }
        }
    }
458373
private void handleRequest(final ChannelHandlerContext ctx, final ByteBuf message) {
    if (isDataMessage(message)) {
      dataEventHandler.onEvent(flowController, message);
    } else if (isControlMessage(message)) {
      controlHandler.onEvent(flowController, message);
    } else if (DcpNoopRequest.is(message)) {
      try {
        ByteBuf buffer = ctx.alloc().buffer();
        DcpNoopResponse.init(buffer);
        MessageUtil.setOpaque(MessageUtil.getOpaque(message), buffer);
        ctx.writeAndFlush(buffer);
      } finally {
        message.release();
      }
    } else {
      try {
        LOGGER.warn("Unknown DCP Message, ignoring. \n{}", MessageUtil.humanize(message));
      } finally {
        message.release();
      }
    }
  }
62929
public void handleMessage(T message) throws Fault {
        if (LOG.isLoggable(Level.FINEST)) {
            LOG.finest("FlowIdProducerOut Interceptor called. isOutbound: "
                    + MessageUtils.isOutbound(message) + ", isRequestor: "
                    + MessageUtils.isRequestor(message));
        }

        if (MessageUtils.isRequestor(message)) {
            handleRequestOut(message);
        } else {
            handleResponseOut(message);
        }

        // Don't write flowId for Oneway responses
        if (isOnewayResponse(message)) {
            return;
        }

        // write FlowId to HTTP and Soap layer
        String flowId = FlowIdHelper.getFlowId(message);
        FlowIdProtocolHeaderCodec.writeFlowId(message, flowId);
        FlowIdSoapCodec.writeFlowId(message, flowId);

    }
123624
public void sendFlushedMessage(SIBUuid8 ignore, SIBUuid12 streamID) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
       SibTr.entry(tc, "sendFlushedMessage", new Object[]{ignore, streamID});
       
    ControlFlushed flushMsg = createControlFlushed(_targetMEUuid, streamID);
    
    // If the destination in a Link add Link specific properties to message
    if( _isLink )
    {
      flushMsg = (ControlFlushed)addLinkProps(flushMsg);
    }
    
    _mpio.sendToMe(_targetMEUuid, SIMPConstants.MSG_HIGH_PRIORITY, flushMsg);   
   
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "sendFlushedMessage");
  }
1000652
protected void handleInboundMessageProperties(final MessageContext messageContext,
                                                  final SoapMessage message) {
        if (messageContext == null) { return; }
        
        final String[] propertyNames = messageContext.getPropertyNames();
        if (propertyNames != null) {
            for (final String propertyName : propertyNames) {
                message.setHeader(propertyName, messageContext.getProperty(propertyName));
            }
        }
    }
558579
@Override
    public void handleMessage(Message message) throws Fault {

        //ignore the messages from SAM Server service itself
        BindingOperationInfo boi = message.getExchange().getBindingOperationInfo();
        if (null != boi){
            String operationName = boi.getName().toString();
            if (SAM_OPERATION.equals(operationName)) {
                return;
            }
        }

        if (isRestWadlRequest(message)) {
            // skip handling REST service WADL requests - temporary fix since will be fixed in CXF soon
            return;
        }

        if (isOnewayResponse(message)) {
            // skip oneway response events
            return;
        }

        // Skip Swagger UI web static, swagger.json and swagger.yaml requests
        if (isSwaggerResourceRequest(message)) {
            return;
        }

        //check MessageID
        checkMessageID(message);

        Event event = mapper.mapToEvent(message);

        if (LOG.isLoggable(Level.FINE)) {
            String id = (event.getMessageInfo() != null) ? event.getMessageInfo().getMessageId() : null;
            LOG.fine("Store event [message_id=" + id + "] in cache.");
        }
        if (null != event) {
            queue.add(event);
        }

        // TESB-20624 skip duplicated event processing
        message.getInterceptorChain().remove(this);
    }
1248639
@Override
    public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent evt) throws Exception {
        if (!(evt instanceof MessageEvent)) {
            ctx.sendDownstream(evt);
            return;
        }

        final MessageEvent e = (MessageEvent) evt;
        final Object originalMessage = e.getMessage();
        final Object encodedMessage = encode(ctx, e.getChannel(), originalMessage);
        if (originalMessage == encodedMessage)
            ctx.sendDownstream(evt);
        else if (encodedMessage != null)
            ctx.sendDownstream(new DownstreamMessageEvent(ctx.getChannel(), e.getFuture(), encodedMessage, e.getRemoteAddress())); // Channels.write(ctx, e.getFuture(), encodedMessage, e.getRemoteAddress());
    }
1149907
public void handleMessage(String message, E_MessageType type, ST max100, Object appID){
		this.count++;

		boolean doMax = false;
		if(this.maxCount!=-1 && this.count>this.maxCount){
			max100.add("name", appID);
			max100.add("number", this.maxCount);
			doMax = true;
		}

		if(this.useSkbConsole==true){
			switch(type){
				case ERROR:
					MessageConsole.conError(message);
					if(doMax==true){
						MessageConsole.conError(max100.render());
					}
					break;
				case INFO:
					MessageConsole.conInfo(message);
					if(doMax==true){
						MessageConsole.conError(max100.render());
					}
					break;
				case WARNING:
					MessageConsole.conWarn(message);
					if(doMax==true){
						MessageConsole.conError(max100.render());
					}
					break;
			}
		}
		else{
			switch(type){
				case ERROR:
					this.logger.error(message);
					if(doMax==true){
						this.logger.error(max100.render());
					}
					break;
				case INFO:
					this.logger.info(message);
					if(doMax==true){
						this.logger.error(max100.render());
					}
					break;
				case WARNING:
					this.logger.warn(message);
					if(doMax==true){
						this.logger.error(max100.render());
					}
					break;
			}
		}
	}
1331308
private void handleApplicationUpdateRequest(SerialMessage incomingMessage) {
		logger.trace("Handle Message Application Update Request");
		int nodeId = incomingMessage.getMessagePayloadByte(1);
		
		logger.trace("Application Update Request from Node " + nodeId);
		UpdateState updateState = UpdateState.getUpdateState(incomingMessage.getMessagePayloadByte(0));
		
		switch (updateState) {
			case NODE_INFO_RECEIVED:
				logger.debug("Application update request, node information received.");			
				int length = incomingMessage.getMessagePayloadByte(2);
				ZWaveNode node = getNode(nodeId);
				
				node.resetResendCount();
				
				for (int i = 6; i < length + 3; i++) {
					int data = incomingMessage.getMessagePayloadByte(i);
					if(data == 0xef )  {
						// TODO: Implement control command classes
						break;
					}
					logger.debug(String.format("Adding command class 0x%02X to the list of supported command classes.", data));
					ZWaveCommandClass commandClass = ZWaveCommandClass.getInstance(data, node, this);
					if (commandClass != null)
						node.addCommandClass(commandClass);
				}
				
				// advance node stage.
				node.advanceNodeStage();
				
				if (incomingMessage.getMessageClass() == this.lastSentMessage.getExpectedReply() && !incomingMessage.isTransActionCanceled()) {
					notifyEventListeners(new ZWaveEvent(ZWaveEventType.TRANSACTION_COMPLETED_EVENT, this.lastSentMessage.getMessageNode(), 1, this.lastSentMessage));
					transactionCompleted.release();
					logger.trace("Released. Transaction completed permit count -> {}", transactionCompleted.availablePermits());
				}
				break;
			case NODE_INFO_REQ_FAILED:
				logger.debug("Application update request, Node Info Request Failed, re-request node info.");
				
				SerialMessage requestInfoMessage = this.lastSentMessage;
				
				if (requestInfoMessage.getMessageClass() != SerialMessage.SerialMessageClass.RequestNodeInfo) {
					logger.warn("Got application update request without node info request, ignoring.");
					return;
				}
					
				if (--requestInfoMessage.attempts >= 0) {
					logger.error("Got Node Info Request Failed while sending this serial message. Requeueing");
					this.enqueue(requestInfoMessage);
				} else
				{
					logger.warn("Node Info Request Failed 3x. Discarding message: {}", lastSentMessage.toString());
				}
				transactionCompleted.release();
				break;
			default:
				logger.warn(String.format("TODO: Implement Application Update Request Handling of %s (0x%02X).", updateState.getLabel(), updateState.getKey()));
		}
	}
545501
protected boolean handleMessageNoLine(LineMessage message) {
        if (LOG.isDebugEnabled())
            LOG.debug("Line {} not found.", hex(message.getLine()));
        switch (message.getType()) {
            case INV:
                send(Message.INVACK((Message.INV) message));
                return true;
            //case CHNGD_OWNR:
            case INVACK:
                return true;
            case GET:
            case GETX:
            case MSG:
                handleNotOwner(message, null);
                return true;
            default:
                return false;
        }
    }
868626
private void manageControl(byte[] messageBytes) throws DevFailed{

        ZMQutils.ControlStructure
                controlStructure = ZMQutils.getInstance().decodeControlBuffer(messageBytes);
        ApiUtil.printTrace("From Control:\n" + controlStructure);
        switch (controlStructure.commandCode) {
            case ZMQutils.ZMQ_END:
                stop = true;
                break;

            case ZMQutils.ZMQ_CONNECT_HEARTBEAT:
                connectIfNotDone(heartbeatSocket, controlStructure);
				//System.out.println("-------> ZMQ_CONNECT_HEARTBEAT: " + controlStructure.eventName);
                heartbeatSocket.subscribe(controlStructure.eventName.getBytes());
                break;


            case ZMQutils.ZMQ_DISCONNECT_HEARTBEAT:
                disconnect(heartbeatSocket, controlStructure.eventName);
                break;

            case ZMQutils.ZMQ_CONNECT_EVENT:
                connectIfNotDone(eventSocket, controlStructure);
                eventSocket.subscribe(controlStructure.eventName.getBytes());
                break;

            case ZMQutils.ZMQ_DISCONNECT_EVENT:
                disconnect(eventSocket, controlStructure.eventName);
                break;

        }
    }
1209315
@SuppressWarnings("checkstyle:illegalcatch")
  private void deliverMessageToTask(final byte[] message) throws TaskMessageHandlerFailure {
    try {
      this.fMessageHandler.get().onNext(new DriverMessageImpl(message));
    } catch (final Throwable throwable) {
      throw new TaskMessageHandlerFailure(throwable);
    }
  }
1141656
public synchronized void handleMessage(ChaincodeMessage message) throws Exception {

		if (message.getType() == ChaincodeMessage.Type.KEEPALIVE){
			logger.debug(String.format("[%s] Recieved KEEPALIVE message, do nothing",
					shortID(message)));
			// Received a keep alive message, we don't do anything with it for now
			// and it does not touch the state machine
				return;
		}

		logger.debug(String.format("[%s]Handling ChaincodeMessage of type: %s(state:%s)",
				shortID(message), message.getType(), fsm.current()));

		if (fsm.eventCannotOccur(message.getType().toString())) {
			String errStr = String.format("[%s]Chaincode handler org.hyperledger.fabric.java.fsm cannot handle message (%s) with payload size (%d) while in state: %s",
					message.getTxid(), message.getType(), message.getPayload().size(), fsm.current());
			ByteString payload = ByteString.copyFromUtf8(errStr);
			ChaincodeMessage errormessage = ChaincodeMessage.newBuilder()
					.setType(ERROR)
					.setPayload(payload)
					.setTxid(message.getTxid())
					.build();
			serialSend(errormessage);
			throw new RuntimeException(errStr);
		}

		// Filter errors to allow NoTransitionError and CanceledError
		// to not propagate for cases where embedded Err == nil.
		try {
			fsm.raiseEvent(message.getType().toString(), message);
		} catch (NoTransitionException e) {
			if (e.error != null) throw e;
			logger.debug("["+ shortID(message)+"]Ignoring NoTransitionError");
		} catch (CancelledException e) {
			if (e.error != null) throw e;
			logger.debug("["+ shortID(message)+"]Ignoring CanceledError");
		}
	}
334161
async def handle_control_event(self, message):
        
        cmd = message[WorkerProtocol.COMMAND]
        logger.debug(__("Manager worker got channel command '{}'.", cmd))

        # Prepare settings for use; Django overlaid by state overlaid by
        # anything immediate in the current packet.
        immediates = {}
        if cmd == WorkerProtocol.COMMUNICATE:
            immediates = message.get(WorkerProtocol.COMMUNICATE_SETTINGS, {}) or {}
        override = self.state.settings_override or {}
        override.update(immediates)
        self.settings_actual = self._marshal_settings()
        self.settings_actual.update(override)

        if cmd == WorkerProtocol.COMMUNICATE:
            try:
                await database_sync_to_async(self._data_scan)(**message[WorkerProtocol.COMMUNICATE_EXTRA])
            except Exception:
                logger.exception("Unknown error occured while processing communicate control command.")
                raise
            finally:
                await self.sync_counter.dec('communicate')

        elif cmd == WorkerProtocol.FINISH:
            try:
                data_id = message[WorkerProtocol.DATA_ID]
                data_location = DataLocation.objects.get(data__id=data_id)
                if not getattr(settings, 'FLOW_MANAGER_KEEP_DATA', False):
                    try:
                        def handle_error(func, path, exc_info):
                            """Handle permission errors while removing data directories."""
                            if isinstance(exc_info[1], PermissionError):
                                os.chmod(path, 0o700)
                                shutil.rmtree(path)

                        # Remove secrets directory, but leave the rest of the runtime directory
                        # intact. Runtime directory will be removed during data purge, when the
                        # data object is removed.
                        secrets_dir = os.path.join(
                            self._get_per_data_dir('RUNTIME_DIR', data_location.subpath),
                            ExecutorFiles.SECRETS_DIR
                        )
                        shutil.rmtree(secrets_dir, onerror=handle_error)
                    except OSError:
                        logger.exception("Manager exception while removing data runtime directory.")

                if message[WorkerProtocol.FINISH_SPAWNED]:
                    await database_sync_to_async(self._data_scan)(**message[WorkerProtocol.FINISH_COMMUNICATE_EXTRA])
            except Exception:
                logger.exception(
                    "Unknown error occured while processing finish control command.",
                    extra={'data_id': data_id}
                )
                raise
            finally:
                await self.sync_counter.dec('executor')

        elif cmd == WorkerProtocol.ABORT:
            await self.sync_counter.dec('executor')

        else:
            logger.error(__("Ignoring unknown manager control command '{}'.", cmd))