Process the input data.

@param event the event
@param channel the channel
359610
@Handler
    public void onInput(Input<CharBuffer> event, IOSubchannel channel) {
        if (!openChannels.contains(channel)) {
            return;
        }
        ManagedBuffer<CharBuffer> out = ManagedBuffer.wrap(
            CharBuffer.wrap(event.data()));
        out.position(out.limit());
        out.flip();
        if (out.backingBuffer().toString().compareToIgnoreCase("/quit") == 0) {
            channel.respond(new Close());
            return;
        }
        channel.respond(Output.fromSource(out, true));
    }
334082
public void processEvent(EventObject event) // for asynchronous
  // (nonblocking) response
  // handling
  {
    synchronized (this) {
    } // wail til message sending has finished (transaction attribute set)

    if (event instanceof ResponseEvent) {
      processResponse((ResponseEvent) event);
    } else if (event instanceof TimeoutEvent) {
      processTimeout((TimeoutEvent) event);
    }

    // note, requests don't come through here, they are handled via the
    // listen/waitFor methods.
  }
1313017
public void write(IRTMPEvent event) {
        if (!connection.isClosed()) {
            final IClientStream stream = connection.getStreamByChannelId(id);
            if (id > 3 && stream == null) {
                log.warn("Non-existant stream for channel id: {}, session: {} discarding: {}", id, connection.getSessionId(), event);
            }
            // if the stream is non-existant, the event will go out with stream id == 0
            final Number streamId = (stream == null) ? 0 : stream.getStreamId();
            write(event, streamId);
        } else {
            log.debug("Connection {} is closed, cannot write to channel: {}", connection.getSessionId(), id);
        }
    }
628665
protected void process(T input) {
        try {
            Optional.ofNullable(this.transformFunction.apply(input))
                    .ifPresent(this.outputPublisher::submit);
        } catch (Exception e) {
            JMExceptionManager.handleException(log, e, "process", input);
        }
    }
432386
public void send(String channel, String message) {
		// CAUSE: Assignment to method parameter
		Pair<Boolean, String> sendValidation = isSendValid(channel, message, null, null);

		//String lMessage = message.replace("\n", "\\n");

		if (sendValidation != null && sendValidation.first) {
			try {
				String messageId = Strings.randomString(8);
				ArrayList<Pair<String, String>> messagesToSend = multiPartMessage(
						message, messageId);
				for (Pair<String, String> messageToSend : messagesToSend) {
					send(channel, messageToSend.second, messageToSend.first,
							sendValidation.second);
				}
			} catch (IOException e) {
				raiseOrtcEvent(EventEnum.OnException, this, e);
			}
		}
	}
932681
private void writeEvent(final ChannelHandlerContext ctx, final Object message) {
        if (isFinishedSent)
            return;
        
        if (message instanceof FinishedMessage)
            isFinishedSent = true;

        Channels.write(ctx, Channels.future(null), message);
    }
1166782
protected boolean handleEvent(AbstractEvent event) throws IOException {
		final Class<?> eventType = event.getClass();

		try {
			// ------------------------------------------------------------
			// Runtime events
			// ------------------------------------------------------------

			// This event is also checked at the (single) input gate to release the respective
			// channel, at which it was received.
			if (eventType == EndOfPartitionEvent.class) {
				return true;
			}
			else if (eventType == EndOfSuperstepEvent.class) {
				return incrementEndOfSuperstepEventAndCheck();
			}

			// ------------------------------------------------------------
			// Task events (user)
			// ------------------------------------------------------------
			else if (event instanceof TaskEvent) {
				taskEventHandler.publish((TaskEvent) event);

				return false;
			}
			else {
				throw new IllegalStateException("Received unexpected event of type " + eventType + " at reader.");
			}
		}
		catch (Throwable t) {
			throw new IOException("Error while handling event of type " + eventType + ": " + t.getMessage(), t);
		}
	}
1073743
protected final Event processEvent() throws IOException {
    while (true) {
      String line;
      try {
        line = readLine();
      } catch (final EOFException ex) {
        if (doneOnce) {
          throw ex;
        }
        doneOnce = true;
        line = "";
      }

      // If the line is empty (a blank line), Dispatch the event, as defined below.
      if (line.isEmpty()) {
        // If the data buffer is an empty string, set the data buffer and the event name buffer to
        // the empty string and abort these steps.
        if (dataBuffer.length() == 0) {
          eventName = "";
          continue;
        }

        // If the event name buffer is not the empty string but is also not a valid NCName,
        // set the data buffer and the event name buffer to the empty string and abort these steps.
        // NOT IMPLEMENTED

        final Event.Builder eventBuilder = new Event.Builder();
        eventBuilder.withEventName(eventName.isEmpty() ? Event.MESSAGE_EVENT : eventName);
        eventBuilder.withData(dataBuffer.toString());

        // Set the data buffer and the event name buffer to the empty string.
        dataBuffer = new StringBuilder();
        eventName = "";

        return eventBuilder.build();
        // If the line starts with a U+003A COLON character (':')
      } else if (line.startsWith(":")) {
        // ignore the line
        // If the line contains a U+003A COLON character (':') character
      } else if (line.contains(":")) {
        // Collect the characters on the line before the first U+003A COLON character (':'),
        // and let field be that string.
        final int colonIdx = line.indexOf(":");
        final String field = line.substring(0, colonIdx);

        // Collect the characters on the line after the first U+003A COLON character (':'),
        // and let value be that string.
        // If value starts with a single U+0020 SPACE character, remove it from value.
        String value = line.substring(colonIdx + 1);
        value = value.startsWith(" ") ? value.substring(1) : value;

        processField(field, value);
      // Otherwise, the string is not empty but does not contain a U+003A COLON character (':')
      // character
      } else {
        processField(line, "");
      }
    }
  }
309587
public void afterResponse(Event event) {
		ChaincodeMessage message = messageHelper(event);
		try {
			sendChannel(message);
			logger.debug(String.format("[%s]Received %s, communicated (state:%s)",
					shortID(message), message.getType(), fsm.current()));
		} catch (Exception e) {
			logger.error(String.format("[%s]error sending %s (state:%s): %s", shortID(message),
					message.getType(), fsm.current(), e));
		}
	}
222206
@Override
    public void handleEvent(Event event) {
        LOG.fine("CustomInfoHandler called");

        if (customInfo != null) {
            event.getCustomInfo().putAll(customInfo);
        }
    }
849690
private synchronized void processProgressEvent(DownstreamChannel channel, WrappedByteBuffer buffer) {
        LOG.entering(CLASS_NAME, "processProgressEvent", buffer);
        try {
            
            // update timestamp that is used to record the timestamp of last received message
            channel.lastMessageTimestamp.set(System.currentTimeMillis());
            channel.buffersToRead.add(buffer);

            WebSocketEmulatedDecoderListener<DownstreamChannel> decoderListener = new WebSocketEmulatedDecoderListener<DownstreamChannel>() {

                @Override
                public void messageDecoded(DownstreamChannel channel, WrappedByteBuffer message) {
                    processMessage(channel, message);
                }

                @Override
                public void messageDecoded(DownstreamChannel channel, String message) {
                    processMessage(channel, message);
                }

                @Override
                public void commandDecoded(DownstreamChannel channel, WrappedByteBuffer command) {
                    int commandByte = command.array()[0];
                    if (commandByte == 0x30 && command.array()[1] == 0x31) { //reconnect
                        // KG-5615: Set flag - but do not reconnect until request has loaded
                        LOG.fine("Reconnect command");
                        channel.reconnecting.set(true);
                    }
                    else if (commandByte == 0x30 && command.array()[1] == 0x32) {
                        channel.closing.set(true);

                        // Cancel the idle timer if running
                        stopIdleTimer(channel);
                        
                        //close frame received
                        int code = CloseCommandMessage.CLOSE_NO_STATUS;
                        String reason = null;
                        command.skip(2); //skip first 2 bytes 0x30, 0x32
                        if (command.hasRemaining()) {
                            code = command.getShort();
                        }
                        if (command.hasRemaining()) {
                            reason = command.getString(Charset.forName("UTF-8"));
                        }
                        CloseCommandMessage message = new CloseCommandMessage(code, reason);
                        listener.commandMessageReceived(channel, message);
                    }
                }

                @Override
                public void pingReceived(DownstreamChannel channel) {
                    listener.pingReceived(channel);
                }
            };

            // Prevent multiple threads from entering
            synchronized (channel.decoder) {
                channel.decoder.decode(channel, in, decoderListener);
            }
        } catch (Exception e) {
            LOG.log(Level.FINE, e.getMessage(), e);
            e.printStackTrace();
            
            listener.downstreamFailed(channel, e);
        }
    }
1072015
void process(EventEnvelope envelope) {
        Object event = getEvent(envelope);
        String serviceName = envelope.getServiceName();
        EventPublishingService<Object, Object> service = eventService.nodeEngine.getService(serviceName);
        Registration registration = getRegistration(envelope, serviceName);
        if (registration == null) {
            return;
        }
        service.dispatchEvent(event, registration.getListener());
    }
992295
@Override
  public void process(final SelectionKey key) {
    try {
      final ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();

      if (serverChannel.isOpen()) {
        final SocketChannel clientChannel = serverChannel.accept();

        if (clientChannel != null) {
          clientChannel.configureBlocking(false);

          clientChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, new Context(clientChannel));
        }
      }
    } catch (Exception e) {
      final String message = "Failed to accept client socket for " + name + " because: " + e.getMessage();
      logger().log(message, e);
      throw new IllegalArgumentException(message);
    }
  }
17314
public static void post(final Event event) {
        if (!isEnable()) {
            return;
        }
        CopyOnWriteArraySet<Subscriber> subscribers = SUBSCRIBER_MAP.get(event.getClass());
        if (CommonUtils.isNotEmpty(subscribers)) {
            for (final Subscriber subscriber : subscribers) {
                if (subscriber.isSync()) {
                    handleEvent(subscriber, event);
                } else { // 异步
                    final RpcInternalContext context = RpcInternalContext.peekContext();
                    final ThreadPoolExecutor asyncThreadPool = AsyncRuntime.getAsyncThreadPool();
                    try {
                        asyncThreadPool.execute(
                            new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        RpcInternalContext.setContext(context);
                                        handleEvent(subscriber, event);
                                    } catch (Exception e) {
                                        RpcInternalContext.removeContext();
                                    }
                                }
                            });
                    } catch (RejectedExecutionException e) {
                        LOGGER
                            .warn("This queue is full when post event to async execute, queue size is " +
                                asyncThreadPool.getQueue().size() +
                                ", please optimize this async thread pool of eventbus.");
                    }
                }
            }
        }
    }
796292
void processEvent(Event e) {
    assert !_processed;
    // Event e = _events[idx];
    if (e.isSend()) {
      _sends.put(e, new ArrayList<TimelineSnapshot.Event>());
      for (Event otherE : _events) {
        if ((otherE != null) && (otherE != e) && (!otherE.equals(e)) && otherE._blocked
          && otherE.match(e)) {
          _edges.put(otherE, e);
          _sends.get(e).add(otherE);
          otherE._blocked = false;
        }
      }
    } else { // look for matching send, otherwise set _blocked
      assert !_edges.containsKey(e);
      int senderIdx = e.packH2O().index();
      if (senderIdx < 0) { // binary search did not find member, should not happen?
        // no possible sender - return and do not block
        Log.warn("no sender found! port = " + e.portPack() + ", ip = " + e.addrPack().toString());
        return;
      }
      Event senderCnd = _events[senderIdx];
      if (senderCnd != null) {
        if (isSenderRecvPair(senderCnd, e)) {
          _edges.put(e, senderCnd.clone());
          _sends.get(senderCnd).add(e);
          return;
        }
        senderCnd = senderCnd.clone();
        while (senderCnd.prev()) {
          if (isSenderRecvPair(senderCnd, e)) {
            _edges.put(e, senderCnd);
            _sends.get(senderCnd).add(e);
            return;
          }
        }
      }
      e._blocked = true;
    }
    assert (e == null) || (e._eventIdx < TimeLine.MAX_EVENTS);
  }
949355
@SuppressWarnings("PMD.UseVarargs")
    public void fire(Event<?> event, Channel[] channels) {
        eventPipeline.add(event, channels);
    }
836516
@MainThread
    private void handleEventPost(Event event) {
        Utils.log(event.getKey(), "Handling posted event");

        int sizeBefore = executionQueue.size();

        scheduleStatusUpdates(event, EventStatus.STARTED);
        scheduleSubscribersInvocation(event);

        if (((EventBase) event).handlersCount == 0) {
            Utils.log(event.getKey(), "No subscribers found");
            // Removing all scheduled STARTED status callbacks
            while (executionQueue.size() > sizeBefore) {
                executionQueue.removeLast();
            }
        } else {
            activeEvents.add(event);
            executeTasks(false);
        }
    }