pushEvent(new IoEvent(IoEventType.MESSAGE_RECEIVED, getSession(), message));
}
private class VmPipeIoProcessor implements IoProcessor<VmPipeSession> {
public void flush(VmPipeSession session) {
WriteRequestQueue queue = session.getWriteRequestQueue0();
if (!session.isClosing()) {
session.getLock().lock();
try {
if (queue.isEmpty(session)) {
return;
}
WriteRequest req;
long currentTime = System.currentTimeMillis();
while ((req = queue.poll(session)) != null) {
Object m = req.getMessage();
pushEvent(new IoEvent(IoEventType.MESSAGE_SENT, session, req), false);
session.getRemoteSession().getFilterChain().fireMessageReceived(getMessageCopy(m));
if (m instanceof IoBuffer) {
session.increaseWrittenBytes0(((IoBuffer) m).remaining(), currentTime);
}
}
} finally {
if (flushEnabled) {
flushEvents();
}
session.getLock().unlock();
}
flushPendingDataQueues(session);
} else {
List<WriteRequest> failedRequests = new ArrayList<WriteRequest>();
WriteRequest req;
while ((req = queue.poll(session)) != null) {
failedRequests.add(req);
}
if (!failedRequests.isEmpty()) {
WriteToClosedSessionException cause = new WriteToClosedSessionException(failedRequests);