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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
<html>
<head>
<title>Apache Qpid : Message API Design</title>
<link rel="stylesheet" href="styles/site.css" type="text/css" />
<META http-equiv="Content-Type" content="text/html; charset=UTF-8">
</head>
<body>
<table class="pagecontent" border="0" cellpadding="0" cellspacing="0" width="100%" bgcolor="#ffffff">
<tr>
<td valign="top" class="pagebody">
<div class="pageheader">
<span class="pagetitle">
Apache Qpid : Message API Design
</span>
</div>
<div class="pagesubheading">
This page last changed on Aug 15, 2007 by <font color="#0050B2">rajith</font>.
</div>
<h1><a name="MessageAPIDesign-MessageAPIDesign"></a>Message API Design</h1>
<p>This document describes the new message API for the restructured client.</p>
<ul>
<li>Sending Messages</li>
<li>Receiving Messages</li>
<li>Message abstraction</li>
<li><a href="http://people.apache.org/~rajith/qpid_docs/client_api/" title="Visit page outside Confluence">Java Doc</a></li>
</ul>
<h2><a name="MessageAPIDesign-SendingMessages"></a>Sending Messages</h2>
<p>The Session class provides the following methods to send messages.</p>
<div class="code" style="border-style: solid; "><div class="codeContent">
<pre class="code-java"><span class="code-keyword">public</span> <span class="code-keyword">interface</span> Session{
.........
<span class="code-comment">//Option1 - <span class="code-keyword">for</span> small messages
</span><span class="code-keyword">public</span> void messageTransfer(<span class="code-object">String</span> destination, Message msg, <span class="code-object">short</span> confirmMode, <span class="code-object">short</span> acquireMode)<span class="code-keyword">throws</span> IOException;
<span class="code-comment">//Option2 - <span class="code-keyword">for</span> large messages
</span><span class="code-keyword">public</span> void messageStream(<span class="code-object">String</span> destination, Message msg, <span class="code-object">short</span> confirmMode, <span class="code-object">short</span> acquireMode)<span class="code-keyword">throws</span> IOException;
<span class="code-comment">//Option3 - can use it with any message size, recomended <span class="code-keyword">for</span> large messages
</span><span class="code-keyword">public</span> void messageTransfer(<span class="code-object">String</span> destination, <span class="code-object">short</span> confirmMode, <span class="code-object">short</span> acquireMode);
<span class="code-keyword">public</span> void headers(Struct... headers);
<span class="code-keyword">public</span> void data(<span class="code-object">byte</span>[] data);
<span class="code-keyword">public</span> void data(ByteBuffer buf);
<span class="code-keyword">public</span> void data(<span class="code-object">String</span> str);
<span class="code-keyword">public</span> void endData();
.........
}</pre>
</div></div>
<h3><a name="MessageAPIDesign-SendingsmallMessages"></a>Sending small Messages </h3>
<p>Option1 provides a convinience method to send small messages.<br/>
You could use the ByteBufferMessage to create small in memory messages (or your own implementation).<br/>
Underneath it maps onto methods defined under option3</p>
<h3><a name="MessageAPIDesign-SendinglargeMessages"></a>Sending large Messages</h3>
<p>You have two options for sending large messages, using either pull style or push style semantics</p>
<h4><a name="MessageAPIDesign-UsingtheSessionclassmethods%28Option3%29"></a>Using the Session class methods (Option3)</h4>
<p>Option3 provides a more natural AMQP style set of methods<br/>
You can stream data using Option3 by pushing your data using one of the data methods defined in the session class.</p>
<h4><a name="MessageAPIDesign-UsingOption2%28pullstyle%29"></a>Using Option2 (pull style)</h4>
<p>The messageStream method will pull data from the message and stream using the methods defined in option3.<br/>
You could use FileMessage or StreamingMessage or your own Message implementation that backs a large data stream.</p>
<ul>
<li>FileMessage takes in a FileInputStream and create a nio FileChannel. It then uses a MappedByteBuffer to map a region of the file when the readData method is invoked. You could specify a chunksize in the constructor to control how much data is mapped each time.</li>
</ul>
<ul>
<li>StreamingMessage takes in a SocketChannel and reads a chunk of data at a time until the SocketChannel is closed. This could be useful when u need to transfer a data stream received from a legacy application or a hardware device. In such cases the StreamingMessage provides a convinient abstraction to stream the data without any intermediate copying.</li>
</ul>
<h2><a name="MessageAPIDesign-ReceivingMessages"></a>Receiving Messages</h2>
<p>To receive messages you can subscribe using the following method</p>
<div class="code" style="border-style: solid; "><div class="codeContent">
<pre class="code-java"><span class="code-keyword">public</span> <span class="code-keyword">interface</span> Session{
.........
<span class="code-keyword">public</span> void messageSubscribe(<span class="code-object">String</span> queue, <span class="code-object">String</span> destination, <span class="code-object">short</span> confirmMode, <span class="code-object">short</span> acquireMode,
MessagePartListener listener, Map<<span class="code-object">String</span>, ?> filter, Option... options);
-----
}</pre>
</div></div>
<p>The API provides support for receiving messages in parts as and when they arrive using the MessagePartListener.<br/>
This enables the user to start consuming the message while it is being streamed.</p>
<div class="code" style="border-style: solid; "><div class="codeContent">
<pre class="code-java"><span class="code-keyword">public</span> <span class="code-keyword">interface</span> MessagePartListener{
<span class="code-keyword">public</span> void messageTransfer(<span class="code-object">long</span> transferId);
<span class="code-keyword">public</span> void messageHeaders(Struct... headers);
<span class="code-keyword">public</span> void data(ByteBuffer src);
<span class="code-keyword">public</span> void messageReceived();
}</pre>
</div></div>
<p>The messageTransfer method signals the start of a transfer and passes the transferId.<br/>
The Transfer Id is used for the following operations defined in the Session API.</p>
<ul>
<li>to Acquire the message (if the message was transfered in no-acquire mode)</li>
<li>to release the message ( if already acquired)</li>
<li>to Reject or Acknowledge the message</li>
</ul>
<p>The data method will be called each time Frame arrives. The messageReceived method will signal the end of the message.</p>
<h3><a name="MessageAPIDesign-Consumingsmallmessages"></a>Consuming small messages</h3>
<p>The API also provides a convinient way for consuming small messages through the MessageListener interface and the MessagePartListenerAdapter.<br/>
The MessagePartListenerAdapter will build the message and will notify the user through MessageListener when the message is complete.</p>
<div class="code" style="border-style: solid; "><div class="codeContent">
<pre class="code-java"><span class="code-keyword">public</span> <span class="code-keyword">interface</span> MessageListener{
<span class="code-keyword">public</span> void onMessage(Message message);
}</pre>
</div></div>
<p>you can use it the following way.</p>
<div class="code" style="border-style: solid; "><div class="codeContent">
<pre class="code-java">.........
MessageListener myMessageListener ....
session.messageSubscribe(....,<span class="code-keyword">new</span> MessagePartListenerDapter(myMessageListener),...);
-----</pre>
</div></div>
<h2><a name="MessageAPIDesign-Messageabstraction"></a>Message abstraction</h2>
<p>Message Interface provides an abstraction for creating messages from different data streams.<br/>
Please read the java doc for a complete description of each method.</p>
<div class="code" style="border-style: solid; "><div class="codeContent">
<pre class="code-java"><span class="code-keyword">public</span> <span class="code-keyword">interface</span> Message{
<span class="code-keyword">public</span> MessageProperties getMessageProperties();
<span class="code-keyword">public</span> DeliveryProperties getDeliveryProperties();
<span class="code-keyword">public</span> void appendData(<span class="code-object">byte</span>[] src) <span class="code-keyword">throws</span> IOException;
<span class="code-keyword">public</span> void appendData(ByteBuffer src) <span class="code-keyword">throws</span> IOException;
<span class="code-keyword">public</span> void readData(<span class="code-object">byte</span>[] target) <span class="code-keyword">throws</span> IOException;
<span class="code-keyword">public</span> ByteBuffer readData() <span class="code-keyword">throws</span> IOException;
<span class="code-keyword">public</span> void clearData();
<span class="code-keyword">public</span> <span class="code-object">long</span> getMessageTransferId();
}</pre>
</div></div>
</td>
</tr>
</table>
<table border="0" cellpadding="0" cellspacing="0" width="100%">
<tr>
<td height="12" background="border/border_bottom.gif"><img src="border/spacer.gif" width="1" height="1" border="0"/></td>
</tr>
<tr>
<td align="center"><font color="grey">Document generated by Confluence on Apr 22, 2008 02:47</font></td>
</tr>
</table>
</body>
</html>
|
