### Question:
ProtocolProcessor { void processConnect(ServerChannel session, ConnectMessage msg) { LOG.info("processConnect for client " + msg.getClientID()); if (msg.getProcotolVersion() != 0x03) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); LOG.info("processConnect sent bad proto ConnAck"); session.write(badProto); session.close(false); return; } if (msg.getClientID() == null || msg.getClientID().length() > 23) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); session.write(okResp); return; } if (m_clientIDs.containsKey(msg.getClientID())) { ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).getSession(); boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION); if (cleanSession) { processRemoveAllSubscriptions(msg.getClientID()); } m_clientIDs.get(msg.getClientID()).getSession().close(false); } ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); m_clientIDs.put(msg.getClientID(), connDescr); int keepAlive = msg.getKeepAlive(); LOG.debug(String.format("Connect with keepAlive %d s", keepAlive)); session.setAttribute(Constants.KEEP_ALIVE, keepAlive); session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); session.setIdleTime(Math.round(keepAlive * 1.5f)); if (msg.isWillFlag()) { AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()]; PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(), msg.isWillRetain(), msg.getClientID(), session); processPublish(pubEvt); } if (msg.isUserFlag()) { String pwd = null; if (msg.isPasswordFlag()) { pwd = msg.getPassword(); } if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); session.write(okResp); return; } } subscriptions.activate(msg.getClientID()); if (msg.isCleanSession()) { processRemoveAllSubscriptions(msg.getClientID()); } ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); LOG.info("processConnect sent OK ConnAck"); session.write(okResp); if (!msg.isCleanSession()) { republishStored(msg.getClientID()); } } ProtocolProcessor();   }### Answer:
@Test public void testHandleConnect_BadProtocol() { connMsg.setProcotolVersion((byte) 0x02); m_processor.processConnect(m_session, connMsg); assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode); }
@Test public void testConnect_badClientID() { connMsg.setClientID("extremely_long_clientID_greater_than_23"); m_processor.processConnect(m_session, connMsg); assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode); }
@Test public void testWill() { connMsg.setClientID("123"); connMsg.setWillFlag(true); connMsg.setWillTopic("topic"); connMsg.setWillMessage("Topic message"); m_processor.processConnect(m_session, connMsg); assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); }