blob: 32f7ccd2e1884e6324ceebe350347a90cc73e5c4 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.messagebus;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* A session supporting receiving and replying to messages. A destination is expected to reply to every message
* received.
*
* @author Simon Thoresen Hult
*/
public final class DestinationSession implements Connectable, MessageHandler {
private final AtomicBoolean destroyed = new AtomicBoolean(false);
private final String name;
private final boolean broadcastName;
private final MessageBus mbus;
private final MessageHandler msgHandler;
/**
* This constructor is package private since only MessageBus is supposed to instantiate it.
*
* @param mbus The message bus that created this instance.
* @param params The parameter object for this session.
*/
DestinationSession(MessageBus mbus, DestinationSessionParams params) {
this.mbus = mbus;
this.name = params.getName();
this.broadcastName = params.getBroadcastName();
this.msgHandler = params.getMessageHandler();
}
/**
* Sets the destroyed flag to true. The very first time this method is called, it cleans up all its dependencies.
* Even if you retain a reference to this object, all of its content is allowed to be garbage collected.
*
* @return True if content existed and was destroyed.
*/
public boolean destroy() {
if (!destroyed.getAndSet(true)) {
close();
return true;
}
return false;
}
/**
* This method unregisters this session from message bus, effectively disabling any more messages from being
* delivered to the message handler. After unregistering, this method calls {@link com.yahoo.messagebus.MessageBus#sync()}
* as to ensure that there are no threads currently entangled in the handler.
*
* This method will deadlock if you call it from the message handler.
*/
public void close() {
mbus.unregisterSession(name, broadcastName);
mbus.sync();
}
/**
* Convenience method for acknowledging a message for its sender.
*
* This is equivalent to:
* <pre>
* Reply ack = new EmptyReply();
* ack.swapState(msg);
* reply(ack);
* </pre>
*
* Messages should be acknowledged when
* <ul>
* <li>this destination has safely and permanently applied the message, or
* <li>an intermediate determines that the purpose of the message is fulfilled without forwarding the message.
* </ul>
*
* @param msg The message to acknowledge back to the sender.
* @see #reply
*/
public void acknowledge(Message msg) {
Reply ack = new EmptyReply();
msg.swapState(ack);
reply(ack);
}
/**
* Sends a reply to a message. The reply will propagate back to the original sender, preferring the same route as it
* used to reach the destination.
*
* @param reply The reply, created from the message this is a reply to.
*/
public void reply(Reply reply) {
ReplyHandler handler = reply.popHandler();
handler.handleReply(reply);
}
/**
* Returns the message handler of this session.
*
* @return The message handler.
*/
public MessageHandler getMessageHandler() {
return msgHandler;
}
/**
* Returns the connection spec string for this session. This returns a combination of the owning message bus' own
* spec string and the name of this session.
*
* @return The connection string.
*/
public String getConnectionSpec() {
return mbus.getConnectionSpec() + "/" + name;
}
/**
* Returns the name of this session.
*
* @return The session name.
*/
public String getName() {
return name;
}
@Override
public void handleMessage(Message msg) {
msgHandler.handleMessage(msg);
}
@Override
public void connect() {
mbus.connect(name, broadcastName);
}
@Override public void disconnect() { close(); }
}
|
