From d18533a88f2ee9ecfd92a94fc3f9734b4b649678 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Fri, 15 Apr 2022 14:19:56 +0300 Subject: [PATCH 1/4] InMemoryStorage extracted --- .../queue/memory/DefaultInMemoryStorage.java | 94 +++++++++++++++++++ .../server/queue/memory/InMemoryStorage.java | 51 +--------- ...t.java => DefaultInMemoryStorageTest.java} | 6 +- 3 files changed, 101 insertions(+), 50 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java rename common/queue/src/test/java/org/thingsboard/server/queue/memory/{InMemoryStorageTest.java => DefaultInMemoryStorageTest.java} (91%) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java new file mode 100644 index 0000000000..f852974dd5 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java @@ -0,0 +1,94 @@ +/** + * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * + * Copyright © 2016-2022 ThingsBoard, Inc. All Rights Reserved. + * + * NOTICE: All information contained herein is, and remains + * the property of ThingsBoard, Inc. and its suppliers, + * if any. The intellectual and technical concepts contained + * herein are proprietary to ThingsBoard, Inc. + * and its suppliers and may be covered by U.S. and Foreign Patents, + * patents in process, and are protected by trade secret or copyright law. + * + * Dissemination of this information or reproduction of this material is strictly forbidden + * unless prior written permission is obtained from COMPANY. + * + * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, + * managers or contractors who have executed Confidentiality and Non-disclosure agreements + * explicitly covering such access. + * + * The copyright notice above does not evidence any actual or intended publication + * or disclosure of this source code, which includes + * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. + * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, + * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT + * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, + * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. + * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION + * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, + * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + */ +package org.thingsboard.server.queue.memory; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.queue.TbQueueMsg; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; + +@Component +@Slf4j +public final class DefaultInMemoryStorage implements InMemoryStorage { + private final ConcurrentHashMap> storage = new ConcurrentHashMap<>(); + + @Override + public void printStats() { + if (log.isDebugEnabled()) { + storage.forEach((topic, queue) -> { + if (queue.size() > 0) { + log.debug("[{}] Queue Size [{}]", topic, queue.size()); + } + }); + } + } + + @Override + public int getLagTotal() { + return storage.values().stream().map(BlockingQueue::size).reduce(0, Integer::sum); + } + + @Override + public boolean put(String topic, TbQueueMsg msg) { + return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).add(msg); + } + + @Override + public List get(String topic) throws InterruptedException { + if (storage.containsKey(topic)) { + List entities; + @SuppressWarnings("unchecked") + T first = (T) storage.get(topic).poll(); + if (first != null) { + entities = new ArrayList<>(); + entities.add(first); + List otherList = new ArrayList<>(); + storage.get(topic).drainTo(otherList, 999); + for (TbQueueMsg other : otherList) { + @SuppressWarnings("unchecked") + T entity = (T) other; + entities.add(entity); + } + } else { + entities = Collections.emptyList(); + } + return entities; + } + return Collections.emptyList(); + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java index 11b97100ee..0fdfebbfff 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java @@ -15,59 +15,18 @@ */ package org.thingsboard.server.queue.memory; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Component; import org.thingsboard.server.queue.TbQueueMsg; -import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.LinkedBlockingQueue; -@Component -@Slf4j -public final class InMemoryStorage { - private final ConcurrentHashMap> storage = new ConcurrentHashMap<>(); +public interface InMemoryStorage { - public void printStats() { - storage.forEach((topic, queue) -> { - if (queue.size() > 0) { - log.debug("[{}] Queue Size [{}]", topic, queue.size()); - } - }); - } + void printStats(); - public int getLagTotal() { - return storage.values().stream().map(BlockingQueue::size).reduce(0, Integer::sum); - } + int getLagTotal(); - public boolean put(String topic, TbQueueMsg msg) { - return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).add(msg); - } + boolean put(String topic, TbQueueMsg msg); - public List get(String topic) throws InterruptedException { - if (storage.containsKey(topic)) { - List entities; - @SuppressWarnings("unchecked") - T first = (T) storage.get(topic).poll(); - if (first != null) { - entities = new ArrayList<>(); - entities.add(first); - List otherList = new ArrayList<>(); - storage.get(topic).drainTo(otherList, 999); - for (TbQueueMsg other : otherList) { - @SuppressWarnings("unchecked") - T entity = (T) other; - entities.add(entity); - } - } else { - entities = Collections.emptyList(); - } - return entities; - } - return Collections.emptyList(); - } + List get(String topic) throws InterruptedException; } diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/memory/InMemoryStorageTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java similarity index 91% rename from common/queue/src/test/java/org/thingsboard/server/queue/memory/InMemoryStorageTest.java rename to common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java index d8911c3fd9..bd4f238447 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/memory/InMemoryStorageTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java @@ -15,17 +15,15 @@ */ package org.thingsboard.server.queue.memory; -import org.junit.After; -import org.junit.Before; import org.junit.Test; import org.thingsboard.server.queue.TbQueueMsg; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; -public class InMemoryStorageTest { +public class DefaultInMemoryStorageTest { - InMemoryStorage storage = new InMemoryStorage(); + InMemoryStorage storage = new DefaultInMemoryStorage(); @Test public void givenStorage_whenGetLagTotal_thenReturnInteger() throws InterruptedException { From b9b4d06376bcc74f77fe72db9770483d1f1e09ff Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Fri, 15 Apr 2022 15:17:57 +0300 Subject: [PATCH 2/4] DefaultInMemoryStorageTest test added on Poll before improvement --- .../memory/DefaultInMemoryStorageTest.java | 27 ++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java index bd4f238447..65d9da1b07 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java @@ -15,12 +15,20 @@ */ package org.thingsboard.server.queue.memory; +import com.google.gson.Gson; +import lombok.extern.slf4j.Slf4j; import org.junit.Test; import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.common.DefaultTbQueueMsg; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; +@Slf4j public class DefaultInMemoryStorageTest { InMemoryStorage storage = new DefaultInMemoryStorage(); @@ -37,4 +45,21 @@ public class DefaultInMemoryStorageTest { storage.get("main"); assertThat(storage.getLagTotal()).isEqualTo(1); } -} \ No newline at end of file + + @Test + public void givenQueue_whenPoll_thenReturnList() throws InterruptedException { + Gson gson = new Gson(); + String topic = "tb_core_notification.tb-node-0"; + List msgs = new ArrayList<>(1001); + for (int i = 0; i < 1001; i++) { + DefaultTbQueueMsg msg = gson.fromJson("{\"key\": \"" + UUID.randomUUID() + "\"}", DefaultTbQueueMsg.class); + msgs.add(msg); + storage.put(topic, msg); + } + + assertThat(storage.getLagTotal()).as("total lag is 1001").isEqualTo(1001); + assertThat(storage.get(topic)).as("poll exactly 1000 msgs").isEqualTo(msgs.subList(0, 1000)); + assertThat(storage.get(topic)).as("poll last 1 message").isEqualTo(msgs.subList(1000, 1001)); + assertThat(storage.getLagTotal()).as("total lag is zero").isEqualTo(0); + } +} From f8a675118292d5e48f21f82251de250a2d5aeeba Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Fri, 15 Apr 2022 16:40:32 +0300 Subject: [PATCH 3/4] InMemoryStorage performance improved. Many test cases added since it is essential piece of code. --- .../queue/memory/DefaultInMemoryStorage.java | 28 ++++---- .../memory/DefaultInMemoryStorageTest.java | 66 ++++++++++++++++--- 2 files changed, 70 insertions(+), 24 deletions(-) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java index f852974dd5..64694a817c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java @@ -67,26 +67,22 @@ public final class DefaultInMemoryStorage implements InMemoryStorage { return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).add(msg); } + @SuppressWarnings("unchecked") @Override public List get(String topic) throws InterruptedException { - if (storage.containsKey(topic)) { - List entities; - @SuppressWarnings("unchecked") - T first = (T) storage.get(topic).poll(); - if (first != null) { - entities = new ArrayList<>(); - entities.add(first); - List otherList = new ArrayList<>(); - storage.get(topic).drainTo(otherList, 999); - for (TbQueueMsg other : otherList) { - @SuppressWarnings("unchecked") - T entity = (T) other; - entities.add(entity); + final BlockingQueue queue = storage.get(topic); + if (queue != null) { + final TbQueueMsg firstMsg = queue.poll(); + if (firstMsg != null) { + final int queueSize = queue.size(); + if (queueSize > 0) { + final List entities = new ArrayList<>(Math.min(queueSize, 999) + 1); + entities.add(firstMsg); + queue.drainTo(entities, 999); + return (List) entities; } - } else { - entities = Collections.emptyList(); + return Collections.singletonList((T) firstMsg); } - return entities; } return Collections.emptyList(); } diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java index 65d9da1b07..b269e55888 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java @@ -30,6 +30,9 @@ import static org.mockito.Mockito.mock; @Slf4j public class DefaultInMemoryStorageTest { + static final int MAX_POLL_SIZE = 1000; + final Gson gson = new Gson(); + final String topic = "tb_core_notification.tb-node-0"; InMemoryStorage storage = new DefaultInMemoryStorage(); @@ -47,19 +50,66 @@ public class DefaultInMemoryStorageTest { } @Test - public void givenQueue_whenPoll_thenReturnList() throws InterruptedException { - Gson gson = new Gson(); - String topic = "tb_core_notification.tb-node-0"; - List msgs = new ArrayList<>(1001); - for (int i = 0; i < 1001; i++) { + public void givenQueueWithMoreThenBatchSize_whenPoll_thenReturnFullListAndSecondList() throws InterruptedException { + List msgs = new ArrayList<>(MAX_POLL_SIZE + 1); + for (int i = 0; i < MAX_POLL_SIZE + 1; i++) { DefaultTbQueueMsg msg = gson.fromJson("{\"key\": \"" + UUID.randomUUID() + "\"}", DefaultTbQueueMsg.class); msgs.add(msg); storage.put(topic, msg); } - assertThat(storage.getLagTotal()).as("total lag is 1001").isEqualTo(1001); - assertThat(storage.get(topic)).as("poll exactly 1000 msgs").isEqualTo(msgs.subList(0, 1000)); - assertThat(storage.get(topic)).as("poll last 1 message").isEqualTo(msgs.subList(1000, 1001)); + assertThat(storage.getLagTotal()).as("total lag is 1001").isEqualTo(MAX_POLL_SIZE + 1); + assertThat(storage.get(topic)).as("poll exactly 1000 msgs").isEqualTo(msgs.subList(0, MAX_POLL_SIZE)); + assertThat(storage.get(topic)).as("poll last 1 message").isEqualTo(msgs.subList(MAX_POLL_SIZE, MAX_POLL_SIZE + 1)); assertThat(storage.getLagTotal()).as("total lag is zero").isEqualTo(0); } + + private void testPollOnce(final int msgCount) throws InterruptedException { + List msgs = new ArrayList<>(msgCount); + for (int i = 0; i < msgCount; i++) { + DefaultTbQueueMsg msg = gson.fromJson("{\"key\": \"" + UUID.randomUUID() + "\"}", DefaultTbQueueMsg.class); + msgs.add(msg); + storage.put(topic, msg); + } + + assertThat(storage.getLagTotal()).as("total lag before poll").isEqualTo(msgCount); + assertThat(storage.get(topic)).as("polled exactly msgs").isEqualTo(msgs.subList(0, msgCount)); + assertThat(storage.getLagTotal()).as("final lag is zero").isEqualTo(0); + } + + @Test + public void givenQueueWithExactBatchSize_whenPoll_thenReturnExactBatchSizeList() throws InterruptedException { + testPollOnce(MAX_POLL_SIZE); + } + + @Test + public void givenQueueWithExactBatchSizeMinusOne_whenPoll_thenReturnCorrectSizeList() throws InterruptedException { + testPollOnce(MAX_POLL_SIZE - 1); + } + + @Test + public void givenQueueWithExactBatchSizeMinusTen_whenPoll_thenReturnCorrectSizeList() throws InterruptedException { + testPollOnce(MAX_POLL_SIZE - 10); + } + + @Test + public void givenQueueEmpty_whenPoll_thenReturnEmptyList() throws InterruptedException { + testPollOnce(0); + } + + @Test + public void givenQueueWithSingleMessage_whenPoll_thenReturnSingletonList() throws InterruptedException { + testPollOnce(1); + } + + @Test + public void givenQueueWithTwoMessages_whenPoll_thenReturnCorrectSizeList() throws InterruptedException { + testPollOnce(2); + } + + @Test + public void givenQueueWithTenMessages_whenPoll_thenReturnCorrectSizeList() throws InterruptedException { + testPollOnce(10); + } + } From 8436e759ee541b8d0436db2b8f45a4e7193ec2a1 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Fri, 15 Apr 2022 16:48:04 +0300 Subject: [PATCH 4/4] DefaultInMemoryStorage fixed license header --- .../queue/memory/DefaultInMemoryStorage.java | 35 ++++++------------- 1 file changed, 10 insertions(+), 25 deletions(-) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java index 64694a817c..6408729877 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorage.java @@ -1,32 +1,17 @@ /** - * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * Copyright © 2016-2022 The Thingsboard Authors * - * Copyright © 2016-2022 ThingsBoard, Inc. All Rights Reserved. + * 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 * - * NOTICE: All information contained herein is, and remains - * the property of ThingsBoard, Inc. and its suppliers, - * if any. The intellectual and technical concepts contained - * herein are proprietary to ThingsBoard, Inc. - * and its suppliers and may be covered by U.S. and Foreign Patents, - * patents in process, and are protected by trade secret or copyright law. + * http://www.apache.org/licenses/LICENSE-2.0 * - * Dissemination of this information or reproduction of this material is strictly forbidden - * unless prior written permission is obtained from COMPANY. - * - * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, - * managers or contractors who have executed Confidentiality and Non-disclosure agreements - * explicitly covering such access. - * - * The copyright notice above does not evidence any actual or intended publication - * or disclosure of this source code, which includes - * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. - * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, - * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT - * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, - * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. - * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION - * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, - * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + * 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.queue.memory;