Skip to content

Commit a18c0f7

Browse files
committed
iluwatar#590 add explanation for Poison Pill
1 parent e34de39 commit a18c0f7

File tree

1 file changed

+244
-2
lines changed

1 file changed

+244
-2
lines changed

poison-pill/README.md

Lines changed: 244 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,250 @@ tags:
1010
---
1111

1212
## Intent
13-
Poison Pill is known predefined data item that allows to provide
14-
graceful shutdown for separate distributed consumption process.
13+
Poison Pill is known predefined data item that allows to provide graceful shutdown for separate distributed consumption
14+
process.
15+
16+
## Explanation
17+
18+
Real world example
19+
20+
> Let's think about a message queue with one producer and one consumer. The producer keeps pushing new messages in the queue and the consumer keeps reading them. Finally when it's time to gracefully shut down the producer sends the poison pill message.
21+
22+
In plain words
23+
24+
> Poison Pill is a known message structure that ends the message exchange.
25+
26+
**Programmatic Example**
27+
28+
Let's define the message structure first.
29+
30+
```java
31+
public interface Message {
32+
33+
Message POISON_PILL = new Message() {
34+
35+
@Override
36+
public void addHeader(Headers header, String value) {
37+
throw poison();
38+
}
39+
40+
@Override
41+
public String getHeader(Headers header) {
42+
throw poison();
43+
}
44+
45+
@Override
46+
public Map<Headers, String> getHeaders() {
47+
throw poison();
48+
}
49+
50+
@Override
51+
public void setBody(String body) {
52+
throw poison();
53+
}
54+
55+
@Override
56+
public String getBody() {
57+
throw poison();
58+
}
59+
60+
private RuntimeException poison() {
61+
return new UnsupportedOperationException("Poison");
62+
}
63+
64+
};
65+
66+
enum Headers {
67+
DATE, SENDER
68+
}
69+
70+
void addHeader(Headers header, String value);
71+
72+
String getHeader(Headers header);
73+
74+
Map<Headers, String> getHeaders();
75+
76+
void setBody(String body);
77+
78+
String getBody();
79+
}
80+
81+
public class SimpleMessage implements Message {
82+
83+
private Map<Headers, String> headers = new HashMap<>();
84+
private String body;
85+
86+
@Override
87+
public void addHeader(Headers header, String value) {
88+
headers.put(header, value);
89+
}
90+
91+
@Override
92+
public String getHeader(Headers header) {
93+
return headers.get(header);
94+
}
95+
96+
@Override
97+
public Map<Headers, String> getHeaders() {
98+
return Collections.unmodifiableMap(headers);
99+
}
100+
101+
@Override
102+
public void setBody(String body) {
103+
this.body = body;
104+
}
105+
106+
@Override
107+
public String getBody() {
108+
return body;
109+
}
110+
}
111+
```
112+
113+
Next we define the types related to the message queue.
114+
115+
```java
116+
public interface MqPublishPoint {
117+
118+
void put(Message msg) throws InterruptedException;
119+
}
120+
121+
public interface MqSubscribePoint {
122+
123+
Message take() throws InterruptedException;
124+
}
125+
126+
public interface MessageQueue extends MqPublishPoint, MqSubscribePoint {
127+
}
128+
129+
public class SimpleMessageQueue implements MessageQueue {
130+
131+
private final BlockingQueue<Message> queue;
132+
133+
public SimpleMessageQueue(int bound) {
134+
queue = new ArrayBlockingQueue<>(bound);
135+
}
136+
137+
@Override
138+
public void put(Message msg) throws InterruptedException {
139+
queue.put(msg);
140+
}
141+
142+
@Override
143+
public Message take() throws InterruptedException {
144+
return queue.take();
145+
}
146+
}
147+
```
148+
149+
Now we need to create the message producer and consumer.
150+
151+
```java
152+
public class Producer {
153+
154+
private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
155+
156+
private final MqPublishPoint queue;
157+
private final String name;
158+
private boolean isStopped;
159+
160+
/**
161+
* Constructor.
162+
*/
163+
public Producer(String name, MqPublishPoint queue) {
164+
this.name = name;
165+
this.queue = queue;
166+
this.isStopped = false;
167+
}
168+
169+
/**
170+
* Send message to queue.
171+
*/
172+
public void send(String body) {
173+
if (isStopped) {
174+
throw new IllegalStateException(String.format(
175+
"Producer %s was stopped and fail to deliver requested message [%s].", body, name));
176+
}
177+
var msg = new SimpleMessage();
178+
msg.addHeader(Headers.DATE, new Date().toString());
179+
msg.addHeader(Headers.SENDER, name);
180+
msg.setBody(body);
181+
182+
try {
183+
queue.put(msg);
184+
} catch (InterruptedException e) {
185+
// allow thread to exit
186+
LOGGER.error("Exception caught.", e);
187+
}
188+
}
189+
190+
/**
191+
* Stop system by sending poison pill.
192+
*/
193+
public void stop() {
194+
isStopped = true;
195+
try {
196+
queue.put(Message.POISON_PILL);
197+
} catch (InterruptedException e) {
198+
// allow thread to exit
199+
LOGGER.error("Exception caught.", e);
200+
}
201+
}
202+
}
203+
204+
public class Consumer {
205+
206+
private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
207+
208+
private final MqSubscribePoint queue;
209+
private final String name;
210+
211+
public Consumer(String name, MqSubscribePoint queue) {
212+
this.name = name;
213+
this.queue = queue;
214+
}
215+
216+
/**
217+
* Consume message.
218+
*/
219+
public void consume() {
220+
while (true) {
221+
try {
222+
var msg = queue.take();
223+
if (Message.POISON_PILL.equals(msg)) {
224+
LOGGER.info("Consumer {} receive request to terminate.", name);
225+
break;
226+
}
227+
var sender = msg.getHeader(Headers.SENDER);
228+
var body = msg.getBody();
229+
LOGGER.info("Message [{}] from [{}] received by [{}]", body, sender, name);
230+
} catch (InterruptedException e) {
231+
// allow thread to exit
232+
LOGGER.error("Exception caught.", e);
233+
return;
234+
}
235+
}
236+
}
237+
}
238+
```
239+
240+
Finally we are ready to present the whole example in action.
241+
242+
```java
243+
var queue = new SimpleMessageQueue(10000);
244+
245+
final var producer = new Producer("PRODUCER_1", queue);
246+
final var consumer = new Consumer("CONSUMER_1", queue);
247+
248+
new Thread(consumer::consume).start();
249+
250+
new Thread(() -> {
251+
producer.send("hand shake");
252+
producer.send("some very important information");
253+
producer.send("bye!");
254+
producer.stop();
255+
}).start();
256+
```
15257

16258
## Class diagram
17259
![alt text](./etc/poison-pill.png "Poison Pill")

0 commit comments

Comments
 (0)