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 it.tidalwave.messagebus.spi.MultiQueue.TopicAndMessage;
30 import lombok.Getter;
31 import lombok.Setter;
32 import lombok.ToString;
33 import lombok.extern.slf4j.Slf4j;
34
35 /***************************************************************************************************************************************************************
36 * An implementation of {@link MessageDelivery} that keeps a queue of messages for each topic, and dispatches them in a round-robin fashion, topic by topic, so
37 * a topic with many messages doesn't starve the others (see {@link MultiQueue}). Messages are dispatched by a number of workers, each of them permanently
38 * running in a thread of the {@link java.util.concurrent.Executor} of the {@link SimpleMessageBus}.
39 * @author Fabrizio Giudici
40 * @since 2.2
41 **************************************************************************************************************************************************************/
42 @Slf4j @ToString(of = "workers")
43 public class RoundRobinAsyncMessageDelivery implements MessageDelivery
44 {
45 @Nonnull
46 private SimpleMessageBus messageBusSupport;
47
48 /***********************************************************************************************************************************************************
49 * The number of workers that dispatch messages.
50 * @return the number of workers
51 **********************************************************************************************************************************************************/
52 @Getter @Setter
53 private int workers = 10;
54
55 private final MultiQueue multiQueue = new MultiQueue();
56
57 /***********************************************************************************************************************************************************
58 * The code run by each worker: it repeatedly removes the next message from the {@link MultiQueue}, waiting if none is available, and dispatches it by means
59 * of the {@link SimpleMessageBus}. It terminates when the thread is interrupted.
60 **********************************************************************************************************************************************************/
61 private final Runnable dispatcher = new Runnable()
62 {
63 @Override
64 public void run()
65 {
66 for (;;)
67 {
68 try
69 {
70 dispatchMessage(multiQueue.remove());
71 }
72 catch (InterruptedException e)
73 {
74 break;
75 }
76 }
77 }
78
79 private <T> void dispatchMessage (@Nonnull final TopicAndMessage<T> tam)
80 {
81 messageBusSupport.dispatchMessage(tam.getTopic(), tam.getMessage());
82 }
83 };
84
85 /***********************************************************************************************************************************************************
86 * {@inheritDoc}
87 **********************************************************************************************************************************************************/
88 @Override
89 public void initialize (@Nonnull final SimpleMessageBus messageBusSupport)
90 {
91 this.messageBusSupport = messageBusSupport;
92 final var executor = this.messageBusSupport.getExecutor();
93
94 for (var i = 0; i < workers; i++)
95 {
96 executor.execute(dispatcher);
97 }
98 }
99
100 /***********************************************************************************************************************************************************
101 * {@inheritDoc}
102 **********************************************************************************************************************************************************/
103 @Override
104 public <T> void deliverMessage (@Nonnull final Class<T> topic, @Nonnull final T message)
105 {
106 multiQueue.add(topic, message);
107 }
108 }