1 /*
2 * *************************************************************************************************************************************************************
3 *
4 * TheseFoolishThings: Miscellaneous utilities
5 * http://tidalwave.it/projects/thesefoolishthings
6 *
7 * Copyright (C) 2009 - 2026 by Tidalwave s.a.s. (http://tidalwave.it)
8 *
9 * *************************************************************************************************************************************************************
10 *
11 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
12 * You may obtain a copy of the License at
13 *
14 * http://www.apache.org/licenses/LICENSE-2.0
15 *
16 * 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
17 * CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
18 *
19 * *************************************************************************************************************************************************************
20 *
21 * git clone https://bitbucket.org/tidalwave/thesefoolishthings-src
22 * git clone https://github.com/tidalwave-it/thesefoolishthings-src
23 *
24 * *************************************************************************************************************************************************************
25 */
26 package it.tidalwave.messagebus.spi;
27
28 import jakarta.annotation.Nonnull;
29 import java.util.ArrayList;
30 import java.util.Comparator;
31 import java.util.List;
32 import java.util.Queue;
33 import java.util.concurrent.ConcurrentNavigableMap;
34 import java.util.concurrent.ConcurrentSkipListMap;
35 import java.util.concurrent.LinkedBlockingQueue;
36 import lombok.Getter;
37 import lombok.RequiredArgsConstructor;
38 import lombok.ToString;
39 import lombok.extern.slf4j.Slf4j;
40
41 /***************************************************************************************************************************************************************
42 * A set of queues of messages, one for each topic. Messages are removed from the queues in a round-robin fashion: each call to {@link #remove()} starts
43 * scanning the queues from the topic following the one of the latest removed message, so a topic with many messages doesn't starve the others. Used by
44 * {@link RoundRobinAsyncMessageDelivery}.
45 * @author Fabrizio Giudici
46 **************************************************************************************************************************************************************/
47 @Slf4j
48 public class MultiQueue
49 {
50 /***********************************************************************************************************************************************************
51 * A pair of a topic and a message.
52 * @param <T> the static type of the topic
53 **********************************************************************************************************************************************************/
54 @RequiredArgsConstructor @Getter @ToString
55 static class TopicAndMessage<T>
56 {
57 @Nonnull
58 private final Class<T> topic;
59
60 @Nonnull
61 private final T message;
62 }
63
64 private final ConcurrentNavigableMap<Class<?>, Queue<?>> queueMapByTopic =
65 new ConcurrentSkipListMap<>(Comparator.comparing(Class::getName));
66
67 private Class<?> latestSentTopic = null;
68
69 /***********************************************************************************************************************************************************
70 * Adds a message of the given topic to this queue and issues a notification.
71 * @param <T> the static type of the message
72 * @param topic the dynamic type of the message
73 * @param message the message
74 **********************************************************************************************************************************************************/
75 public synchronized <T> void add (@Nonnull final Class<T> topic, @Nonnull final T message)
76 {
77 getQueue(topic).add(message);
78 notifyAll();
79 }
80
81 /***********************************************************************************************************************************************************
82 * Removes and returns the next pair (topic, message) from the queue. Blocks until one is available.
83 * @param <T> the static type of the topic
84 * @return the topic and message
85 * @throws InterruptedException if interrupted while waiting
86 **********************************************************************************************************************************************************/
87 @Nonnull
88 public synchronized <T> TopicAndMessage<T> remove()
89 throws InterruptedException
90 {
91 for (;;)
92 {
93 for (final var topic : reorderedTopics())
94 {
95 final var queue = queueMapByTopic.get(topic);
96 final var message = queue.poll();
97
98 if (message != null)
99 {
100 latestSentTopic = topic;
101
102 if (log.isTraceEnabled())
103 {
104 log.trace("stats {}", stats());
105 }
106
107 return new TopicAndMessage<>((Class<T>)topic, (T)message);
108 }
109 }
110
111 if (log.isTraceEnabled())
112 {
113 log.trace("all queues empty; stats {}", stats());
114 }
115
116 wait();
117 }
118 }
119
120 /***********************************************************************************************************************************************************
121 * Returns the list of topics reordered, so it starts just after latestSentTopic and wraps around.
122 **********************************************************************************************************************************************************/
123 @Nonnull
124 private List<Class<?>> reorderedTopics()
125 {
126 final var keySet = queueMapByTopic.navigableKeySet();
127 final List<Class<?>> scanSet = new ArrayList<>();
128
129 if (latestSentTopic == null)
130 {
131 scanSet.addAll(keySet);
132 }
133 else
134 {
135 scanSet.addAll(keySet.subSet(latestSentTopic, false, keySet.last(), true));
136 scanSet.addAll(keySet.subSet(keySet.first(), true, latestSentTopic, true));
137 }
138
139 return scanSet;
140 }
141
142 /***********************************************************************************************************************************************************
143 * Returns a description of the state of the queues, for logging: for each topic, the number of messages in its queue; the topic of the latest removed
144 * message is marked with {@code X}.
145 * @return the description
146 **********************************************************************************************************************************************************/
147 private synchronized String stats()
148 {
149 final var b = new StringBuilder();
150 var separator = "";
151
152 for (final var e : queueMapByTopic.entrySet())
153 {
154 b.append(separator).append(String.format("%s[%s]: %d",
155 e.getKey().getSimpleName(), e.getKey().equals(latestSentTopic) ? "X" : " ", e.getValue().size()));
156 separator = ", ";
157 }
158
159 return b.toString();
160 }
161
162 /***********************************************************************************************************************************************************
163 * Returns the queue associated to a given topic. The queue is created if the topic is new.
164 * @param topic the topic
165 * @return the queue
166 **********************************************************************************************************************************************************/
167 @Nonnull
168 private synchronized <T> Queue<T> getQueue (@Nonnull final Class<T> topic)
169 {
170 // TODO Java 8 would make this easier
171 var queue = (Queue<T>)queueMapByTopic.get(topic);
172
173 if (queue == null)
174 {
175 queue = new LinkedBlockingQueue<>();
176 queueMapByTopic.put(topic, queue);
177 }
178
179 return queue;
180 }
181 }