Vladyslav_Prykhodko
5 years ago
70 changed files with 895 additions and 294 deletions
@ -0,0 +1,27 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.transport.lwm2m.bootstrap.secure; |
||||
|
|
||||
|
import org.eclipse.leshan.server.bootstrap.BootstrapConfigStore; |
||||
|
import org.eclipse.leshan.server.bootstrap.BootstrapConfigurationStoreAdapter; |
||||
|
|
||||
|
public class LwM2MInMemoryBootstrapConfigurationAdapter extends BootstrapConfigurationStoreAdapter { |
||||
|
|
||||
|
public LwM2MInMemoryBootstrapConfigurationAdapter(BootstrapConfigStore store) { |
||||
|
super(store); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,221 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.transport.mqtt; |
||||
|
|
||||
|
import io.netty.buffer.ByteBuf; |
||||
|
import io.netty.buffer.EmptyByteBuf; |
||||
|
import io.netty.buffer.PooledByteBufAllocator; |
||||
|
import io.netty.channel.ChannelHandlerContext; |
||||
|
import io.netty.handler.codec.mqtt.MqttConnectMessage; |
||||
|
import io.netty.handler.codec.mqtt.MqttConnectPayload; |
||||
|
import io.netty.handler.codec.mqtt.MqttConnectVariableHeader; |
||||
|
import io.netty.handler.codec.mqtt.MqttFixedHeader; |
||||
|
import io.netty.handler.codec.mqtt.MqttMessage; |
||||
|
import io.netty.handler.codec.mqtt.MqttMessageType; |
||||
|
import io.netty.handler.codec.mqtt.MqttPublishMessage; |
||||
|
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader; |
||||
|
import io.netty.handler.codec.mqtt.MqttQoS; |
||||
|
import io.netty.handler.ssl.SslHandler; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.junit.After; |
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.junit.runner.RunWith; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
||||
|
|
||||
|
import java.net.InetSocketAddress; |
||||
|
import java.nio.charset.StandardCharsets; |
||||
|
import java.util.List; |
||||
|
import java.util.concurrent.CountDownLatch; |
||||
|
import java.util.concurrent.ExecutorService; |
||||
|
import java.util.concurrent.Executors; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
import java.util.stream.Collectors; |
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static org.hamcrest.MatcherAssert.assertThat; |
||||
|
import static org.hamcrest.Matchers.contains; |
||||
|
import static org.hamcrest.Matchers.empty; |
||||
|
import static org.hamcrest.Matchers.greaterThan; |
||||
|
import static org.hamcrest.Matchers.is; |
||||
|
import static org.junit.Assert.fail; |
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.BDDMockito.willDoNothing; |
||||
|
import static org.mockito.BDDMockito.willReturn; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
import static org.mockito.Mockito.spy; |
||||
|
import static org.mockito.Mockito.times; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
|
||||
|
@Slf4j |
||||
|
@RunWith(MockitoJUnitRunner.class) |
||||
|
public class MqttTransportHandlerTest { |
||||
|
|
||||
|
public static final int MSG_QUEUE_LIMIT = 10; |
||||
|
public static final InetSocketAddress IP_ADDR = new InetSocketAddress("127.0.0.1", 9876); |
||||
|
public static final int TIMEOUT = 30; |
||||
|
|
||||
|
@Mock |
||||
|
MqttTransportContext context; |
||||
|
@Mock |
||||
|
SslHandler sslHandler; |
||||
|
@Mock |
||||
|
ChannelHandlerContext ctx; |
||||
|
|
||||
|
AtomicInteger packedId = new AtomicInteger(); |
||||
|
ExecutorService executor; |
||||
|
MqttTransportHandler handler; |
||||
|
|
||||
|
@Before |
||||
|
public void setUp() throws Exception { |
||||
|
willReturn(MSG_QUEUE_LIMIT).given(context).getMessageQueueSizePerDeviceLimit(); |
||||
|
|
||||
|
handler = spy(new MqttTransportHandler(context, sslHandler)); |
||||
|
willReturn(IP_ADDR).given(handler).getAddress(any()); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void tearDown() { |
||||
|
if (executor != null) { |
||||
|
executor.shutdownNow(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
MqttConnectMessage getMqttConnectMessage() { |
||||
|
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, true, MqttQoS.AT_LEAST_ONCE, false, 123); |
||||
|
MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader("device", packedId.incrementAndGet(), true, true, true, 1, true, false, 60); |
||||
|
MqttConnectPayload payload = new MqttConnectPayload("clientId", "topic", "message".getBytes(StandardCharsets.UTF_8), "username", "password".getBytes(StandardCharsets.UTF_8)); |
||||
|
return new MqttConnectMessage(mqttFixedHeader, variableHeader, payload); |
||||
|
} |
||||
|
|
||||
|
MqttPublishMessage getMqttPublishMessage() { |
||||
|
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, true, MqttQoS.AT_LEAST_ONCE, false, 123); |
||||
|
MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader("v1/gateway/telemetry", packedId.incrementAndGet()); |
||||
|
ByteBuf payload = new EmptyByteBuf(new PooledByteBufAllocator()); |
||||
|
return new MqttPublishMessage(mqttFixedHeader, variableHeader, payload); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenMessageWithoutFixedHeader_whenProcessMqttMsg_thenProcessDisconnect() { |
||||
|
MqttFixedHeader mqttFixedHeader = null; |
||||
|
MqttMessage msg = new MqttMessage(mqttFixedHeader); |
||||
|
willDoNothing().given(handler).processDisconnect(ctx); |
||||
|
|
||||
|
handler.processMqttMsg(ctx, msg); |
||||
|
|
||||
|
assertThat(handler.address, is(IP_ADDR)); |
||||
|
verify(handler, times(1)).processDisconnect(ctx); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenMqttConnectMessage_whenProcessMqttMsg_thenProcessConnect() { |
||||
|
MqttConnectMessage msg = getMqttConnectMessage(); |
||||
|
willDoNothing().given(handler).processConnect(ctx, msg); |
||||
|
|
||||
|
handler.processMqttMsg(ctx, msg); |
||||
|
|
||||
|
assertThat(handler.address, is(IP_ADDR)); |
||||
|
assertThat(handler.deviceSessionCtx.getChannel(), is(ctx)); |
||||
|
verify(handler, never()).processDisconnect(any()); |
||||
|
verify(handler, times(1)).processConnect(ctx, msg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenQueueLimit_whenEnqueueRegularSessionMsgOverLimit_thenOK() { |
||||
|
List<MqttPublishMessage> messages = Stream.generate(this::getMqttPublishMessage).limit(MSG_QUEUE_LIMIT).collect(Collectors.toList()); |
||||
|
messages.forEach(msg -> handler.enqueueRegularSessionMsg(ctx, msg)); |
||||
|
assertThat(handler.deviceSessionCtx.getMsgQueueSize().get(), is(MSG_QUEUE_LIMIT)); |
||||
|
assertThat(handler.deviceSessionCtx.getMsgQueue(), contains(messages.toArray())); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenQueueLimit_whenEnqueueRegularSessionMsgOverLimit_thenCtxClose() { |
||||
|
final int limit = MSG_QUEUE_LIMIT + 1; |
||||
|
willDoNothing().given(handler).processMsgQueue(ctx); |
||||
|
List<MqttPublishMessage> messages = Stream.generate(this::getMqttPublishMessage).limit(limit).collect(Collectors.toList()); |
||||
|
|
||||
|
messages.forEach((msg) -> handler.enqueueRegularSessionMsg(ctx, msg)); |
||||
|
|
||||
|
assertThat(handler.deviceSessionCtx.getMsgQueueSize().get(), is(limit)); |
||||
|
verify(handler, times(limit)).enqueueRegularSessionMsg(any(), any()); |
||||
|
verify(handler, times(MSG_QUEUE_LIMIT)).processMsgQueue(any()); |
||||
|
verify(ctx, times(1)).close(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenMqttConnectMessageAndPublishImmediately_whenProcessMqttMsg_thenEnqueueRegularSessionMsg() { |
||||
|
givenMqttConnectMessage_whenProcessMqttMsg_thenProcessConnect(); |
||||
|
|
||||
|
List<MqttPublishMessage> messages = Stream.generate(this::getMqttPublishMessage).limit(MSG_QUEUE_LIMIT).collect(Collectors.toList()); |
||||
|
|
||||
|
messages.forEach((msg) -> handler.processMqttMsg(ctx, msg)); |
||||
|
|
||||
|
assertThat(handler.address, is(IP_ADDR)); |
||||
|
assertThat(handler.deviceSessionCtx.getChannel(), is(ctx)); |
||||
|
assertThat(handler.deviceSessionCtx.isConnected(), is(false)); |
||||
|
assertThat(handler.deviceSessionCtx.getMsgQueueSize().get(), is(MSG_QUEUE_LIMIT)); |
||||
|
assertThat(handler.deviceSessionCtx.getMsgQueue(), contains(messages.toArray())); |
||||
|
verify(handler, never()).processDisconnect(any()); |
||||
|
verify(handler, times(1)).processConnect(any(), any()); |
||||
|
verify(handler, times(MSG_QUEUE_LIMIT)).enqueueRegularSessionMsg(any(), any()); |
||||
|
verify(handler, never()).processRegularSessionMsg(any(), any()); |
||||
|
messages.forEach((msg) -> verify(handler, times(1)).enqueueRegularSessionMsg(ctx, msg)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenMessageQueue_whenProcessMqttMsgConcurrently_thenEnqueueRegularSessionMsg() throws InterruptedException { |
||||
|
//given
|
||||
|
assertThat(handler.deviceSessionCtx.isConnected(), is(false)); |
||||
|
assertThat(MSG_QUEUE_LIMIT, greaterThan(2)); |
||||
|
List<MqttPublishMessage> messages = Stream.generate(this::getMqttPublishMessage).limit(MSG_QUEUE_LIMIT).collect(Collectors.toList()); |
||||
|
messages.forEach((msg) -> handler.enqueueRegularSessionMsg(ctx, msg)); |
||||
|
willDoNothing().given(handler).processRegularSessionMsg(any(), any()); |
||||
|
executor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName(getClass().getName())); |
||||
|
|
||||
|
CountDownLatch readyLatch = new CountDownLatch(MSG_QUEUE_LIMIT); |
||||
|
CountDownLatch startLatch = new CountDownLatch(1); |
||||
|
CountDownLatch finishLatch = new CountDownLatch(MSG_QUEUE_LIMIT); |
||||
|
|
||||
|
Stream.iterate(0, i -> i + 1).limit(MSG_QUEUE_LIMIT).forEach(x -> |
||||
|
executor.submit(() -> { |
||||
|
try { |
||||
|
readyLatch.countDown(); |
||||
|
assertThat(startLatch.await(TIMEOUT, TimeUnit.SECONDS), is(true)); |
||||
|
handler.processMsgQueue(ctx); |
||||
|
finishLatch.countDown(); |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to run processMsgQueue", e); |
||||
|
fail("Failed to run processMsgQueue"); |
||||
|
} |
||||
|
})); |
||||
|
|
||||
|
//when
|
||||
|
assertThat(readyLatch.await(TIMEOUT, TimeUnit.SECONDS), is(true)); |
||||
|
handler.deviceSessionCtx.setConnected(true); |
||||
|
startLatch.countDown(); |
||||
|
assertThat(finishLatch.await(TIMEOUT, TimeUnit.SECONDS), is(true)); |
||||
|
|
||||
|
//then
|
||||
|
assertThat(handler.deviceSessionCtx.getMsgQueueSize().get(), is(0)); |
||||
|
assertThat(handler.deviceSessionCtx.getMsgQueue(), empty()); |
||||
|
verify(handler, times(MSG_QUEUE_LIMIT)).processRegularSessionMsg(any(), any()); |
||||
|
messages.forEach((msg) -> verify(handler, times(1)).processRegularSessionMsg(ctx, msg)); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,61 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.transport.mqtt.session; |
||||
|
|
||||
|
import org.junit.Test; |
||||
|
|
||||
|
import java.util.WeakHashMap; |
||||
|
import java.util.concurrent.ConcurrentMap; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.locks.Lock; |
||||
|
import java.util.concurrent.locks.ReentrantLock; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
import static org.junit.Assert.assertTrue; |
||||
|
import static org.mockito.BDDMockito.willCallRealMethod; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
|
||||
|
public class GatewaySessionHandlerTest { |
||||
|
|
||||
|
@Test |
||||
|
public void givenWeakHashMap_WhenGC_thenMapIsEmpty() { |
||||
|
WeakHashMap<String, Lock> map = new WeakHashMap<>(); |
||||
|
|
||||
|
String deviceName = new String("device"); //constants are static and doesn't affected by GC, so use new instead
|
||||
|
map.put(deviceName, new ReentrantLock()); |
||||
|
assertTrue(map.containsKey(deviceName)); |
||||
|
|
||||
|
deviceName = null; |
||||
|
System.gc(); |
||||
|
|
||||
|
await().atMost(10, TimeUnit.SECONDS).until(() -> !map.containsKey("device")); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenConcurrentReferenceHashMap_WhenGC_thenMapIsEmpty() { |
||||
|
GatewaySessionHandler gsh = mock(GatewaySessionHandler.class); |
||||
|
willCallRealMethod().given(gsh).createWeakMap(); |
||||
|
|
||||
|
ConcurrentMap<String, Lock> map = gsh.createWeakMap(); |
||||
|
map.put("device", new ReentrantLock()); |
||||
|
assertTrue(map.containsKey("device")); |
||||
|
|
||||
|
System.gc(); |
||||
|
|
||||
|
await().atMost(10, TimeUnit.SECONDS).until(() -> !map.containsKey("device")); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue