You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2025/05/20 21:40:39 UTC
(qpid-protonj2) branch main updated (fba2e778 -> 925d3bf3)
This is an automated email from the ASF dual-hosted git repository.
tabish pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
from fba2e778 PROTON-2894 Add an optimized peekByte to buffer APIs
new 20ef704a PROTON-2887 Remove some unused code from DeliveryQueue
new 925d3bf3 PROTON-2887 Fix some build warnings and other minor cleanups
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pom.xml | 3 +-
.../client/impl/ClientNextReceiverSelector.java | 3 -
.../qpid/protonj2/client/util/DeliveryQueue.java | 19 ---
.../protonj2/client/util/FifoDeliveryQueue.java | 153 +++++++--------------
.../client/util/ReconnectLocationPool.java | 24 ++--
.../qpid/protonj2/engine/util/UnsettledMap.java | 4 +-
6 files changed, 65 insertions(+), 141 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
(qpid-protonj2) 02/02: PROTON-2887 Fix some build warnings and other minor cleanups
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
commit 925d3bf372044b9299b039eb3d026f1f58a15042
Author: Timothy Bish <ta...@gmail.com>
AuthorDate: Tue May 20 17:32:33 2025 -0400
PROTON-2887 Fix some build warnings and other minor cleanups
Fixes some warnings and cleans up some code.
---
pom.xml | 3 ++-
.../client/impl/ClientNextReceiverSelector.java | 3 ---
.../client/util/ReconnectLocationPool.java | 24 +++++++++-------------
.../qpid/protonj2/engine/util/UnsettledMap.java | 4 +---
4 files changed, 13 insertions(+), 21 deletions(-)
diff --git a/pom.xml b/pom.xml
index 174c391d..4e7addf5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -62,6 +62,8 @@
<netty-transport-native-epoll-classifier>linux-x86_64</netty-transport-native-epoll-classifier>
<netty-transport-native-kqueue-classifier>osx-x86_64</netty-transport-native-kqueue-classifier>
+ <jacoco-config></jacoco-config>
+
<!-- surefire forked jvm arguments -->
<argLine>-Xmx2g -enableassertions ${jacoco-config}</argLine>
@@ -219,7 +221,6 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
- <optimize>true</optimize>
<showDeprecation>true</showDeprecation>
<showWarnings>true</showWarnings>
</configuration>
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java
index 8d8e8619..a6f04b5c 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java
@@ -162,7 +162,6 @@ final class ClientNextReceiverSelector {
return result != null ? result : selectFirstAvailable();
}
- @SuppressWarnings("resource")
private ClientReceiver selectFirstAvailable() {
return session.getProtonSession().receivers().stream()
.filter((r) -> r.getLinkedResource() instanceof ClientReceiver &&
@@ -172,7 +171,6 @@ final class ClientNextReceiverSelector {
.orElse(null);
}
- @SuppressWarnings("resource")
private ClientReceiver selectLargestBacklog() {
return session.getProtonSession().receivers().stream()
.filter((r) -> r.getLinkedResource() instanceof ClientReceiver &&
@@ -182,7 +180,6 @@ final class ClientNextReceiverSelector {
.orElse(null);
}
- @SuppressWarnings("resource")
private ClientReceiver selectSmallestBacklog() {
return session.getProtonSession().receivers().stream()
.filter((r) -> r.getLinkedResource() instanceof ClientReceiver &&
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/ReconnectLocationPool.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/ReconnectLocationPool.java
index d4d5d3d0..c86fa637 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/ReconnectLocationPool.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/ReconnectLocationPool.java
@@ -48,9 +48,7 @@ public class ReconnectLocationPool {
this.entries = new LinkedList<ReconnectLocation>();
if (backups != null) {
- for (ReconnectLocation entry : backups) {
- this.add(entry);
- }
+ backups.forEach(entry -> add(entry));
}
}
@@ -80,7 +78,7 @@ public class ReconnectLocationPool {
* @return the next entry that should be used for a connection attempt.
*/
public ReconnectLocation getNext() {
- ReconnectLocation next = null;
+ ReconnectLocation next = null;
synchronized (entries) {
if (!entries.isEmpty()) {
next = entries.removeFirst();
@@ -125,9 +123,7 @@ public class ReconnectLocationPool {
public void addAll(List<ReconnectLocation> additions) {
if (additions != null && !additions.isEmpty()) {
synchronized (entries) {
- for (ReconnectLocation entry : additions) {
- add(entry);
- }
+ additions.forEach(entry -> add(entry));
}
}
}
@@ -161,13 +157,13 @@ public class ReconnectLocationPool {
*/
public boolean remove(ReconnectLocation entry) {
if (entry != null) {
- synchronized (entries) {
- for (ReconnectLocation candidate : entries) {
- if (compareEntries(entry, candidate)) {
- return entries.remove(candidate);
- }
- }
- }
+ synchronized (entries) {
+ for (ReconnectLocation candidate : entries) {
+ if (compareEntries(entry, candidate)) {
+ return entries.remove(candidate);
+ }
+ }
+ }
}
return false;
diff --git a/protonj2/src/main/java/org/apache/qpid/protonj2/engine/util/UnsettledMap.java b/protonj2/src/main/java/org/apache/qpid/protonj2/engine/util/UnsettledMap.java
index b4f49e0d..27fc14fa 100644
--- a/protonj2/src/main/java/org/apache/qpid/protonj2/engine/util/UnsettledMap.java
+++ b/protonj2/src/main/java/org/apache/qpid/protonj2/engine/util/UnsettledMap.java
@@ -93,9 +93,7 @@ public class UnsettledMap<Delivery> implements Map<UnsignedInteger, Delivery> {
@Override
public void putAll(Map<? extends UnsignedInteger, ? extends Delivery> source) {
- for (Entry<? extends UnsignedInteger, ? extends Delivery> entry : source.entrySet()) {
- put(entry.getKey(), entry.getValue());
- }
+ source.entrySet().forEach(entry -> put(entry.getKey(), entry.getValue()));
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
(qpid-protonj2) 01/02: PROTON-2887 Remove some unused code from DeliveryQueue
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
commit 20ef704a6def7119629a45d033a3605a57c42ac7
Author: Timothy Bish <ta...@gmail.com>
AuthorDate: Fri May 2 11:49:46 2025 -0400
PROTON-2887 Remove some unused code from DeliveryQueue
Remove some older unused code in the message passing delivery queue and
clean up a bit.
---
.../qpid/protonj2/client/util/DeliveryQueue.java | 19 ---
.../protonj2/client/util/FifoDeliveryQueue.java | 153 +++++++--------------
2 files changed, 52 insertions(+), 120 deletions(-)
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/DeliveryQueue.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/DeliveryQueue.java
index 70dcb9d5..b5a551ca 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/DeliveryQueue.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/DeliveryQueue.java
@@ -33,14 +33,6 @@ public interface DeliveryQueue {
*/
void enqueue(ClientDelivery delivery);
- /**
- * Adds the given {@link Delivery} to the front of the queue.
- *
- * @param delivery
- * The in-bound Delivery to enqueue.
- */
- void enqueueFirst(ClientDelivery delivery);
-
/**
* Used to get an {@link Delivery}. The amount of time this method blocks is based on the timeout value
* that is supplied to it.
@@ -89,22 +81,11 @@ public interface DeliveryQueue {
*/
void stop();
- /**
- * Closes the Delivery Queue. No Delivery can be added or removed from the Queue
- * once it has entered the closed state.
- */
- void close();
-
/**
* @return true if the Queue is not in the stopped or closed state.
*/
boolean isRunning();
- /**
- * @return true if the Queue has been closed.
- */
- boolean isClosed();
-
/**
* @return true if there are no deliveries in the queue.
*/
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/FifoDeliveryQueue.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/FifoDeliveryQueue.java
index 2bbb8f40..a68ad9a7 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/FifoDeliveryQueue.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/util/FifoDeliveryQueue.java
@@ -18,7 +18,6 @@ package org.apache.qpid.protonj2.client.util;
import java.util.ArrayDeque;
import java.util.Deque;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.qpid.protonj2.client.Delivery;
import org.apache.qpid.protonj2.client.impl.ClientDelivery;
@@ -28,19 +27,11 @@ import org.apache.qpid.protonj2.client.impl.ClientDelivery;
*/
public final class FifoDeliveryQueue implements DeliveryQueue {
- private static final AtomicIntegerFieldUpdater<FifoDeliveryQueue> STATE_FIELD_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(FifoDeliveryQueue.class, "state");
-
- private static final int CLOSED = 0;
- private static final int STOPPED = 1;
- private static final int RUNNING = 2;
-
- private volatile int state = STOPPED;
+ private final Deque<ClientDelivery> queue;
+ private volatile boolean started;
private int waiters = 0;
- private final Deque<ClientDelivery> queue;
-
/**
* Creates a new first in / first out message queue with the given queue depth
*
@@ -52,136 +43,96 @@ public final class FifoDeliveryQueue implements DeliveryQueue {
}
@Override
- public void enqueueFirst(ClientDelivery envelope) {
- synchronized (queue) {
- queue.addFirst(envelope);
- if (waiters > 0) {
- queue.notify();
- }
- }
- }
-
- @Override
- public void enqueue(ClientDelivery envelope) {
- synchronized (queue) {
- queue.addLast(envelope);
- if (waiters > 0) {
- queue.notify();
- }
+ public synchronized void enqueue(ClientDelivery envelope) {
+ queue.addLast(envelope);
+ if (waiters > 0) {
+ notify();
}
}
@Override
- public ClientDelivery dequeue(long timeout) throws InterruptedException {
- synchronized (queue) {
- // Wait until the receiver is ready to deliver messages.
- while (timeout != 0 && isRunning() && queue.isEmpty()) {
- if (timeout == -1) {
- waiters++;
- try {
- queue.wait();
- } finally {
- waiters--;
- }
- } else {
- long start = System.currentTimeMillis();
- waiters++;
- try {
- queue.wait(timeout);
- } finally {
- waiters--;
- }
- timeout = Math.max(timeout + start - System.currentTimeMillis(), 0);
+ public synchronized ClientDelivery dequeue(long timeout) throws InterruptedException {
+ // Wait until the receiver is ready to deliver messages.
+ while (queue.isEmpty() && timeout != 0 && started) {
+ if (timeout == -1) {
+ waiters++;
+ try {
+ wait();
+ } finally {
+ waiters--;
}
+ } else {
+ long start = System.currentTimeMillis();
+ waiters++;
+ try {
+ wait(timeout);
+ } finally {
+ waiters--;
+ }
+ timeout = Math.max(timeout + start - System.currentTimeMillis(), 0);
}
+ }
- if (!isRunning()) {
- return null;
- }
-
+ if (started) {
return queue.pollFirst();
+ } else {
+ return null;
}
}
@Override
- public ClientDelivery dequeueNoWait() {
- synchronized (queue) {
- if (!isRunning()) {
- return null;
- }
-
+ public synchronized ClientDelivery dequeueNoWait() {
+ if (started) {
return queue.pollFirst();
+ } else {
+ return null;
}
}
@Override
- public void start() {
- if (STATE_FIELD_UPDATER.compareAndSet(this, STOPPED, RUNNING)) {
- synchronized (queue) {
- if (waiters > 0) {
- queue.notifyAll();
- }
- }
- }
- }
+ public synchronized void start() {
+ if (!started) {
+ started = true;
- @Override
- public void stop() {
- if (STATE_FIELD_UPDATER.compareAndSet(this, RUNNING, STOPPED)) {
- synchronized (queue) {
- if (waiters > 0) {
- queue.notifyAll();
- }
+ if (waiters > 0) {
+ notifyAll();
}
}
}
@Override
- public void close() {
- if (STATE_FIELD_UPDATER.getAndSet(this, CLOSED) > CLOSED) {
- synchronized (queue) {
- if (waiters > 0) {
- queue.notifyAll();
- }
+ public synchronized void stop() {
+ if (started) {
+ started = false;
+
+ if (waiters > 0) {
+ notifyAll();
}
}
}
@Override
public boolean isRunning() {
- return state == RUNNING;
+ return started;
}
@Override
- public boolean isClosed() {
- return state == CLOSED;
+ public synchronized boolean isEmpty() {
+ return queue.isEmpty();
}
@Override
- public boolean isEmpty() {
- synchronized (queue) {
- return queue.isEmpty();
- }
+ public synchronized int size() {
+ return queue.size();
}
@Override
- public int size() {
- synchronized (queue) {
- return queue.size();
- }
+ public synchronized void clear() {
+ queue.clear();
}
@Override
- public void clear() {
- synchronized (queue) {
- queue.clear();
- }
- }
-
- @Override
- public String toString() {
- synchronized (queue) {
- return queue.toString();
- }
+ public synchronized String toString() {
+ return queue.toString();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org