Java 类io.netty.handler.codec.mqtt.MqttConnAckMessage 实例源码

@Test
public void testCONNECTION_REFUSED_SERVER_UNAVAILABLE() throws Exception {
    ServiceChecker prev = Plugins.INSTANCE.put(ServiceChecker.class, new ServiceChecker() {
        @Override
        public Plugin clone() {
            return this;
        }

        @Override
        public boolean isServiceAvailable() {
            return false;
        }
    });

    MqttConnAckMessage ret = executeNormalChannelRead0(TestUtil.newClientId(), true, null);

    Assert.assertEquals(ret.variableHeader().connectReturnCode(),
            MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);

    Plugins.INSTANCE.put(ServiceChecker.class, prev);
}

@Test
public void testCONNECTION_REFUSED_NOT_AUTHORIZED() throws Exception {
    Authorizer prev = Plugins.INSTANCE.put(Authorizer.class, new Authorizer() {
        @Override
        public Plugin clone() {
            return this;
        }

        @Override
        public boolean isAuthorized(String clientId, String username) {
            return false;
        }
    });

    MqttConnAckMessage ret = executeNormalChannelRead0(TestUtil.newClientId(), true, null);

    Assert.assertEquals(ret.variableHeader().connectReturnCode(),
            MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);

    Plugins.INSTANCE.put(Authorizer.class, prev);
}

private MqttConnAckMessage executeNormalChannelRead0(String clientId, boolean cleanSession, ChannelId channelId)
        throws Exception {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
            10);
    MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader("MQTT", 4, true, true, true, 0, true,
            cleanSession, 60);
    MqttConnectPayload payload = new MqttConnectPayload(clientId, "willtopic", "willmessage", "username",
            "password");

    MqttConnectMessage msg = new MqttConnectMessage(fixedHeader, variableHeader, payload);

    ChannelId cid = channelId == null ? TestUtil.newChannelId(clientId, false) : channelId;

    EmbeddedChannel channel = new EmbeddedChannel(cid, new ConnectReceiver());

    channel.writeInbound(msg);

    return channel.readOutbound();
}

public static MqttConnAckMessage connack(MqttConnectReturnCode returnCode, boolean sessionPresent) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false,
            2);
    MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);

    return new MqttConnAckMessage(fixedHeader, variableHeader);
}

@Test
public void testCleanSessionWithoutClientIdReturnFalse() throws Exception {
    Settings.INSTANCE.setProperty("mqttserver.acceptEmptyClientId", "false");
    MqttConnAckMessage ret = executeNormalChannelRead0("", true, null);

    Assert.assertEquals(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED,
            ret.variableHeader().connectReturnCode());
    Settings.INSTANCE.setProperty("mqttserver.acceptEmptyClientId", "true");
}

@Test
public void nonCleanSession() throws Exception {
    String clientId = TestUtil.newClientId();

    MqttConnAckMessage ret = executeNormalChannelRead0(clientId, false, null);

    Assert.assertEquals(ret.variableHeader().connectReturnCode(), MqttConnectReturnCode.CONNECTION_ACCEPTED);
}

static void assertEqualsConnAck(String msg, MqttConnectReturnCode expectedCode, Object connAck) {
    assertTrue("connAck is not an instance of ConnAckMessage", connAck instanceof MqttConnAckMessage);
    MqttConnAckMessage connAckMsg = (MqttConnAckMessage) connAck;

    if (msg == null)
        assertEquals(expectedCode, connAckMsg.variableHeader().connectReturnCode());
    else
        assertEquals(msg, expectedCode, connAckMsg.variableHeader().connectReturnCode());
}

private MqttConnAckMessage createMqttConnAckMsg(MqttConnectReturnCode returnCode) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(CONNACK, false, AT_MOST_ONCE, false, 0);
  MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, true);
  return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}

private MqttConnAckMessage createMqttConnAckMsg(MqttConnectReturnCode returnCode) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(CONNACK, false, AT_MOST_ONCE, false, 0);
  MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, true);
  return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}

public MqttConnectReturnCode connect() throws InterruptedException {

        Class<? extends SocketChannel> socketChannelClass;

        if (Literals.NETTY_EPOLL.equals(Settings.INSTANCE.nettyTransportMode())) {
            group = new EpollEventLoopGroup(1, new DefaultThreadFactory("client"));
            socketChannelClass = EpollSocketChannel.class;
        }
        else {
            group = new NioEventLoopGroup(1, new DefaultThreadFactory("client"));
            socketChannelClass = NioSocketChannel.class;
        }

        bootstrap.group(group).channel(socketChannelClass).handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                if ("mqtts".equalsIgnoreCase(uri.getScheme())) {
                    SslContext sslCtx = SslContextBuilder.forClient().trustManager(trustManagerFactory).build();

                    ch.pipeline().addLast(sslCtx.newHandler(ch.alloc(), uri.getHost(), uri.getPort()));
                }

                ch.pipeline().addLast(MqttDecoder.class.getName(), new MqttDecoder());
                ch.pipeline().addLast(MqttEncoder.class.getName(), MqttEncoder.INSTANCE);
                ch.pipeline().addLast(MqttPacketReceiver.class.getName(),
                        new MqttPacketReceiver(MqttClient.this, receiver, sharedObject));
            }
        });

        channel = bootstrap.connect(uri.getHost(), uri.getPort()).sync().channel();

        normalizeMessage(options.will());
        send(MqttMessageFactory.connect(options));

        synchronized (sharedObject.locker()) {
            int timeout = Settings.INSTANCE.getInt("mqttclient.responseTimeoutSeconds", 15);

            sharedObject.locker().wait(timeout * 1000);
        }
        if (sharedObject.receivedMessage() == null) { return null; }

        return ((MqttConnAckMessage) sharedObject.receivedMessage()).variableHeader().connectReturnCode();
    }

@Test
public void testDefaultChannelRead0() throws Exception {
    MqttConnAckMessage ret = executeNormalChannelRead0(TestUtil.newClientId(), true, null);

    Assert.assertEquals(ret.variableHeader().connectReturnCode(), MqttConnectReturnCode.CONNECTION_ACCEPTED);
}

@Test
public void testCleanSessionWithoutClientIdReturnTrue() throws Exception {
    MqttConnAckMessage ret = executeNormalChannelRead0("", true, null);

    Assert.assertEquals(MqttConnectReturnCode.CONNECTION_ACCEPTED, ret.variableHeader().connectReturnCode());
}

private static MqttConnAckMessage createConnAckMessage(MqttConnectReturnCode code) {
    MqttFixedHeader mqttFixedHeader =
            new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
    MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(code, true);
    return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}

void sendConnack(MqttConnectReturnCode returnCode) {
   MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
   MqttConnAckVariableHeader varHeader = new MqttConnAckVariableHeader(returnCode, true);
   MqttConnAckMessage message = new MqttConnAckMessage(fixedHeader, varHeader);
   sendToClient(message);
}

@Test
public void cleanSessionOnSameClientIdSession() throws Exception {
    String clientId = TestUtil.newClientId();

    executeNormalChannelRead0(clientId, true, null);

    MqttConnAckMessage ret = executeNormalChannelRead0(clientId, true, null);

    Assert.assertEquals(ret.variableHeader().connectReturnCode(), MqttConnectReturnCode.CONNECTION_ACCEPTED);
}