-
Notifications
You must be signed in to change notification settings - Fork 22
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Upgrade Pulsar client (3.2.2) and improve JMS priority on consumer side
- Loading branch information
1 parent
5c009c6
commit f0929a1
Showing
7 changed files
with
302 additions
and
34 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
155 changes: 155 additions & 0 deletions
155
pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PriorityGrowableArrayBlockingQueue.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,155 @@ | ||
| package com.datastax.oss.pulsar.jms; | ||
|
|
||
| import org.apache.pulsar.client.api.Message; | ||
| import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue; | ||
|
|
||
| import javax.annotation.Nullable; | ||
| import java.util.*; | ||
| import java.util.concurrent.PriorityBlockingQueue; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.function.Consumer; | ||
|
|
||
| public class PriorityGrowableArrayBlockingQueue extends GrowableArrayBlockingQueue<Message> { | ||
|
|
||
| static int getPriority(Message m) { | ||
| String jmsPriority = m.getProperty("JMSPriority"); | ||
| if (jmsPriority == null || jmsPriority.isEmpty()) { | ||
| return PulsarMessage.DEFAULT_PRIORITY; | ||
| } | ||
| try { | ||
| return Integer.parseInt(jmsPriority); | ||
| } catch (NumberFormatException err) { | ||
| return PulsarMessage.DEFAULT_PRIORITY; | ||
| } | ||
| } | ||
|
|
||
| private final PriorityBlockingQueue<Message> queue; | ||
| private final AtomicBoolean terminated = new AtomicBoolean(false); | ||
| public PriorityGrowableArrayBlockingQueue() { | ||
| this(10); | ||
| } | ||
|
|
||
| public PriorityGrowableArrayBlockingQueue(int initialCapacity) { | ||
| queue = new PriorityBlockingQueue<>(initialCapacity, new Comparator<Message>() { | ||
| @Override | ||
| public int compare(Message o1, Message o2) { | ||
| int priority1 = getPriority(o1); | ||
| int priority2 = getPriority(o2); | ||
| return Integer.compare(priority2, priority1); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| @Override | ||
| public Message remove() { | ||
| return queue.remove(); | ||
| } | ||
|
|
||
| @Override | ||
| public Message poll() { | ||
| return queue.poll(); | ||
| } | ||
|
|
||
| @Override | ||
| public Message element() { | ||
| return queue.element(); | ||
| } | ||
|
|
||
| @Override | ||
| public Message peek() { | ||
| return queue.peek(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean offer(Message e) { | ||
| return queue.offer(e); | ||
| } | ||
|
|
||
| @Override | ||
| public void put(Message e) { | ||
| queue.put(e); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean add(Message e) { | ||
| return queue.add(e); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean offer(Message e, long timeout, TimeUnit unit) { | ||
| return queue.offer(e, timeout, unit); | ||
| } | ||
|
|
||
| @Override | ||
| public Message take() throws InterruptedException { | ||
| return queue.take(); | ||
| } | ||
|
|
||
| @Override | ||
| public Message poll(long timeout, TimeUnit unit) throws InterruptedException { | ||
| return queue.poll(timeout, unit); | ||
| } | ||
|
|
||
| @Override | ||
| public int remainingCapacity() { | ||
| return queue.remainingCapacity(); | ||
| } | ||
|
|
||
| @Override | ||
| public int drainTo(Collection<? super Message> c) { | ||
| return queue.drainTo(c); | ||
| } | ||
|
|
||
| @Override | ||
| public int drainTo(Collection<? super Message> c, int maxElements) { | ||
| return queue.drainTo(c, maxElements); | ||
| } | ||
|
|
||
| @Override | ||
| public void clear() { | ||
| queue.clear(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean remove(Object o) { | ||
| return queue.remove(o); | ||
| } | ||
|
|
||
| @Override | ||
| public int size() { | ||
| return queue.size(); | ||
| } | ||
|
|
||
| @Override | ||
| public Iterator<Message> iterator() { | ||
| return queue.iterator(); | ||
| } | ||
|
|
||
| @Override | ||
| public List<Message> toList() { | ||
| List<Message> list = new ArrayList<>(size()); | ||
| forEach(list::add); | ||
| return list; | ||
| } | ||
|
|
||
| @Override | ||
| public void forEach(Consumer<? super Message> action) { | ||
| queue.forEach(action); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return queue.toString(); | ||
| } | ||
|
|
||
| @Override | ||
| public void terminate(@Nullable Consumer<Message> itemAfterTerminatedHandler) { | ||
| terminated.set(true); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isTerminated() { | ||
| return terminated.get(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.