This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new c8d9df802 [ISSUE #3827] Changed Access Modifiers for MessageQueue
items (#3987)
c8d9df802 is described below
commit c8d9df80268821906d7466a99830ad356708258c
Author: Ishaan bajpai <[email protected]>
AuthorDate: Tue May 30 17:04:27 2023 +0530
[ISSUE #3827] Changed Access Modifiers for MessageQueue items (#3987)
* Issue #3827
Changed Access modifier of `items[]`
Added getters for the same
Changed Breaking tests
* Renamed reentrant lock inside method scopes
---
.../storage/standalone/broker/MessageQueue.java | 42 ++++++++++++----------
.../standalone/broker/MessageQueueTest.java | 14 ++++----
.../broker/task/HistoryMessageClearTest.java | 2 +-
3 files changed, 31 insertions(+), 27 deletions(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
index dfa60f29d..c887bf693 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
@@ -29,7 +29,7 @@ import com.google.common.base.Preconditions;
*/
public class MessageQueue {
- public MessageEntity[] items;
+ private final MessageEntity[] items;
private int takeIndex;
@@ -65,15 +65,15 @@ public class MessageQueue {
*/
public void put(MessageEntity messageEntity) throws InterruptedException {
Preconditions.checkNotNull(messageEntity);
- ReentrantLock lock = this.lock;
- lock.lockInterruptibly();
+ ReentrantLock reentrantLock = this.lock;
+ reentrantLock.lockInterruptibly();
try {
while (count == items.length) {
notFull.await();
}
enqueue(messageEntity);
} finally {
- lock.unlock();
+ reentrantLock.unlock();
}
}
@@ -84,15 +84,15 @@ public class MessageQueue {
* @throws InterruptedException
*/
public MessageEntity take() throws InterruptedException {
- ReentrantLock lock = this.lock;
- lock.lockInterruptibly();
+ ReentrantLock reentrantLock = this.lock;
+ reentrantLock.lockInterruptibly();
try {
while (count == 0) {
notEmpty.await();
}
return dequeue();
} finally {
- lock.unlock();
+ reentrantLock.unlock();
}
}
@@ -102,12 +102,12 @@ public class MessageQueue {
* @return MessageEntity
*/
public MessageEntity peek() {
- ReentrantLock lock = this.lock;
- lock.lock();
+ ReentrantLock reentrantLock = this.lock;
+ reentrantLock.lock();
try {
return itemAt(takeIndex);
} finally {
- lock.unlock();
+ reentrantLock.unlock();
}
}
@@ -126,8 +126,8 @@ public class MessageQueue {
* @return MessageEntity
*/
public MessageEntity getTail() {
- ReentrantLock lock = this.lock;
- lock.lock();
+ ReentrantLock reentrantLock = this.lock;
+ reentrantLock.lock();
try {
if (count == 0) {
return null;
@@ -138,7 +138,7 @@ public class MessageQueue {
}
return itemAt(tailIndex);
} finally {
- lock.unlock();
+ reentrantLock.unlock();
}
}
@@ -149,8 +149,8 @@ public class MessageQueue {
* @return MessageEntity
*/
public MessageEntity getByOffset(long offset) {
- ReentrantLock lock = this.lock;
- lock.lock();
+ ReentrantLock reentrantLock = this.lock;
+ reentrantLock.lock();
try {
MessageEntity head = getHead();
if (head == null) {
@@ -170,13 +170,13 @@ public class MessageQueue {
}
return itemAt(offsetIndex);
} finally {
- lock.unlock();
+ reentrantLock.unlock();
}
}
public void removeHead() {
- ReentrantLock lock = this.lock;
- lock.lock();
+ ReentrantLock reentrantLock = this.lock;
+ reentrantLock.lock();
try {
if (count == 0) {
return;
@@ -187,7 +187,7 @@ public class MessageQueue {
}
notFull.signalAll();
} finally {
- lock.unlock();
+ reentrantLock.unlock();
}
}
@@ -225,4 +225,8 @@ public class MessageQueue {
public int getPutIndex() {
return putIndex;
}
+
+ public MessageEntity[] getItems() {
+ return items;
+ }
}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/MessageQueueTest.java
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/MessageQueueTest.java
index c6ef39a59..d78c39c6c 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/MessageQueueTest.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/MessageQueueTest.java
@@ -43,42 +43,42 @@ public class MessageQueueTest {
@Test
public void testPut() {
-
Assert.assertTrue(Arrays.stream(messageQueue.items).findAny().isPresent());
+
Assert.assertTrue(Arrays.stream(messageQueue.getItems()).findAny().isPresent());
}
@Test
public void testTake() throws InterruptedException {
MessageEntity takeMessage = messageQueue.take();
Assert.assertNotNull(takeMessage);
-
Assert.assertTrue(Arrays.stream(messageQueue.items).findAny().isPresent());
+
Assert.assertTrue(Arrays.stream(messageQueue.getItems()).findAny().isPresent());
}
@Test
public void testPeek() {
MessageEntity peekMessage = messageQueue.peek();
Assert.assertNotNull(peekMessage);
-
Assert.assertTrue(Arrays.stream(messageQueue.items).findAny().isPresent());
+
Assert.assertTrue(Arrays.stream(messageQueue.getItems()).findAny().isPresent());
}
@Test
public void testGetHead() {
MessageEntity headMessage = messageQueue.getHead();
Assert.assertNotNull(headMessage);
-
Assert.assertTrue(Arrays.stream(messageQueue.items).findAny().isPresent());
+
Assert.assertTrue(Arrays.stream(messageQueue.getItems()).findAny().isPresent());
}
@Test
public void testGetTail() {
MessageEntity tailMessage = messageQueue.getHead();
Assert.assertNotNull(tailMessage);
-
Assert.assertTrue(Arrays.stream(messageQueue.items).findAny().isPresent());
+
Assert.assertTrue(Arrays.stream(messageQueue.getItems()).findAny().isPresent());
}
@Test
public void testGetByOffset() {
MessageEntity offSetMessageEntity =
messageQueue.getByOffset(DEFAULT_OFFSET);
Assert.assertNotNull(offSetMessageEntity);
-
Assert.assertTrue(Arrays.stream(messageQueue.items).findAny().isPresent());
+
Assert.assertTrue(Arrays.stream(messageQueue.getItems()).findAny().isPresent());
Assert.assertEquals(DEFAULT_OFFSET, offSetMessageEntity.getOffset());
}
@@ -91,7 +91,7 @@ public class MessageQueueTest {
@Test
public void testRemoveHead() {
messageQueue.removeHead();
-
Assert.assertTrue(Arrays.stream(messageQueue.items).anyMatch(Objects::isNull));
+
Assert.assertTrue(Arrays.stream(messageQueue.getItems()).anyMatch(Objects::isNull));
}
@Test
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/task/HistoryMessageClearTest.java
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/task/HistoryMessageClearTest.java
index 052e74d34..9e98a19a8 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/task/HistoryMessageClearTest.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/test/java/org/apache/eventmesh/storage/standalone/broker/task/HistoryMessageClearTest.java
@@ -53,6 +53,6 @@ public class HistoryMessageClearTest {
public void testClearMessages() {
historyMessageClear.clearMessages();
MessageQueue updatedMessageQueue = messageContainer.get(new
TopicMetadata(TEST_TOPIC));
-
Assert.assertTrue(Arrays.stream(updatedMessageQueue.items).allMatch(Objects::isNull));
+
Assert.assertTrue(Arrays.stream(updatedMessageQueue.getItems()).allMatch(Objects::isNull));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
