[DIRMINA-1121] Mina v2.1.1 -v2.1.3 are unable to receive writeRequest message(packet loss) in concurrent access

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Not A Problem
• Affects Version/s: 2.1.1, 2.1.2, 2.1.3
• Component/s: Filter
• Assignee: Jonathan Valliere
• Reporter: li-libo
• Created: 11/Jan/20 12:41
• Updated: 08/Jun/20 18:05
• Resolved: 08/Jun/20 18:05

Description

Mina v2.1.1-v2.1.3 are unable to receive writeRequest message(packet loss) in concurrent access, the following code is bug and my bug fix in ProtocolCodecFilter # filterWrite method

Note: The bug 2 is the cause of packet loss!

/**

• {@inheritDoc}

  */
  @Override
  public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception
  { Object message = writeRequest.getMessage(); // Bypass the encoding if the message is contained in a IoBuffer, // as it has already been encoded before if ((message instanceof IoBuffer) || (message instanceof FileRegion))

  { nextFilter.filterWrite(session, writeRequest); return; }

// Get the encoder in the session
ProtocolEncoder encoder = factory.getEncoder(session);

// bug 1: The encoderOut is public that may cause duplicated message to sent.
// ProtocolEncoderOutput encoderOut = getEncoderOut(session, nextFilter, writeRequest);

// bug 1 fix:
ProtocolEncoderOutput encoderOut = new ProtocolEncoderOutputImpl(session, nextFilter, writeRequest);

if (encoder == null)

{ throw new ProtocolEncoderException("The encoder is null for the session " + session); }

try {
// Now we can try to encode the response
encoder.encode(session, message, encoderOut);

// Send it directly
Queue<Object> bufferQueue = ((AbstractProtocolEncoderOutput) encoderOut).getMessageQueue();

// Write all the encoded messages now
while (!bufferQueue.isEmpty()) {
Object encodedMessage = bufferQueue.poll();

if (encodedMessage == null)

{ break; }

// Flush only when the buffer has remaining.
if (!(encodedMessage instanceof IoBuffer) || ((IoBuffer) encodedMessage).hasRemaining()) {
// major bug 2: The same writeRequest's message may be rewritten when the code enters while loop repeatedly, it will lead to the current message  is lost.

// writeRequest.setMessage(encodedMessage);
// nextFilter.filterWrite(session, writeRequest);
// bug 2 fix:
DefaultWriteRequest defaultWriteRequest = new DefaultWriteRequest(writeRequest.getOriginalMessage(),
writeRequest.getFuture(), writeRequest.getDestination());
defaultWriteRequest.setMessage(encodedMessage);
nextFilter.filterWrite(session, defaultWriteRequest);
}
}
} catch (Exception e) {
ProtocolEncoderException pee;

// Generate the correct exception
if (e instanceof ProtocolEncoderException)

{ pee = (ProtocolEncoderException) e; }

else

{ pee = new ProtocolEncoderException(e); }

throw pee;
}
}