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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
|
From 5803aadd3f209eba1ffbb2cf7bf16778019dbee1 Mon Sep 17 00:00:00 2001
From: fredoboulo <fredoboulo@users.noreply.github.com>
Date: Fri, 23 Feb 2018 23:55:57 +0100
Subject: [PATCH] Fix #524 : V1 and V2 protocol downgrades handle received data
in handshake buffer
This patch is upstream pull request, see:
https://gihub.com/zeromq/jeromq/pull/527.
It is merged on commit c2afa9c, and we can drop it on the
0.4.4 release.
---
src/main/java/zmq/io/StreamEngine.java | 21 ++++++++++--
src/test/java/zmq/io/AbstractProtocolVersion.java | 41 +++++++++++++----------
src/test/java/zmq/io/V0ProtocolTest.java | 12 +++++++
src/test/java/zmq/io/V1ProtocolTest.java | 16 +++++++--
src/test/java/zmq/io/V2ProtocolTest.java | 16 +++++++--
5 files changed, 81 insertions(+), 25 deletions(-)
diff --git a/src/main/java/zmq/io/StreamEngine.java b/src/main/java/zmq/io/StreamEngine.java
index b8933c92..fe2f2d8d 100644
--- a/src/main/java/zmq/io/StreamEngine.java
+++ b/src/main/java/zmq/io/StreamEngine.java
@@ -816,9 +816,7 @@ private boolean handshake()
assert (bufferSize == headerSize);
// Make sure the decoder sees the data we have already received.
- greetingRecv.flip();
- inpos = greetingRecv;
- insize = greetingRecv.limit();
+ decodeDataAfterHandshake(0);
// To allow for interoperability with peers that do not forward
// their subscriptions, we inject a phantom subscription message
@@ -846,6 +844,8 @@ else if (greetingRecv.get(revisionPos) == Protocol.V1.revision) {
}
encoder = new V1Encoder(errno, Config.OUT_BATCH_SIZE.getValue());
decoder = new V1Decoder(errno, Config.IN_BATCH_SIZE.getValue(), options.maxMsgSize, options.allocator);
+
+ decodeDataAfterHandshake(V2_GREETING_SIZE);
}
else if (greetingRecv.get(revisionPos) == Protocol.V2.revision) {
// ZMTP/2.0 framing.
@@ -859,6 +859,8 @@ else if (greetingRecv.get(revisionPos) == Protocol.V2.revision) {
}
encoder = new V2Encoder(errno, Config.OUT_BATCH_SIZE.getValue());
decoder = new V2Decoder(errno, Config.IN_BATCH_SIZE.getValue(), options.maxMsgSize, options.allocator);
+
+ decodeDataAfterHandshake(V2_GREETING_SIZE);
}
else {
zmtpVersion = Protocol.V3;
@@ -904,6 +906,19 @@ else if (greetingRecv.get(revisionPos) == Protocol.V2.revision) {
return true;
}
+ private void decodeDataAfterHandshake(int greetingSize)
+ {
+ final int pos = greetingRecv.position();
+ if (pos > greetingSize) {
+ // data is present after handshake
+ greetingRecv.position(greetingSize).limit(pos);
+
+ // Make sure the decoder sees this extra data.
+ inpos = greetingRecv;
+ insize = greetingRecv.remaining();
+ }
+ }
+
private Msg identityMsg()
{
Msg msg = new Msg(options.identitySize);
diff --git a/src/test/java/zmq/io/AbstractProtocolVersion.java b/src/test/java/zmq/io/AbstractProtocolVersion.java
index e60db403..aa06b4a7 100644
--- a/src/test/java/zmq/io/AbstractProtocolVersion.java
+++ b/src/test/java/zmq/io/AbstractProtocolVersion.java
@@ -18,15 +18,18 @@
import zmq.SocketBase;
import zmq.ZError;
import zmq.ZMQ;
+import zmq.ZMQ.Event;
import zmq.util.Utils;
public abstract class AbstractProtocolVersion
{
+ protected static final int REPETITIONS = 1000;
+
static class SocketMonitor extends Thread
{
- private final Ctx ctx;
- private final String monitorAddr;
- private final List<ZMQ.Event> events = new ArrayList<>();
+ private final Ctx ctx;
+ private final String monitorAddr;
+ private final ZMQ.Event[] events = new ZMQ.Event[1];
public SocketMonitor(Ctx ctx, String monitorAddr)
{
@@ -41,15 +44,15 @@ public void run()
boolean rc = s.connect(monitorAddr);
assertThat(rc, is(true));
// Only some of the exceptional events could fire
- while (true) {
- ZMQ.Event event = ZMQ.Event.read(s);
- if (event == null && s.errno() == ZError.ETERM) {
- break;
- }
- assertThat(event, notNullValue());
-
- events.add(event);
+
+ ZMQ.Event event = ZMQ.Event.read(s);
+ if (event == null && s.errno() == ZError.ETERM) {
+ s.close();
+ return;
}
+ assertThat(event, notNullValue());
+
+ events[0] = event;
s.close();
}
}
@@ -69,11 +72,12 @@ public void run()
boolean rc = ZMQ.setSocketOption(receiver, ZMQ.ZMQ_LINGER, 0);
assertThat(rc, is(true));
- SocketMonitor monitor = new SocketMonitor(ctx, "inproc://monitor");
- monitor.start();
rc = ZMQ.monitorSocket(receiver, "inproc://monitor", ZMQ.ZMQ_EVENT_HANDSHAKE_PROTOCOL);
assertThat(rc, is(true));
+ SocketMonitor monitor = new SocketMonitor(ctx, "inproc://monitor");
+ monitor.start();
+
rc = ZMQ.bind(receiver, host);
assertThat(rc, is(true));
@@ -81,17 +85,18 @@ public void run()
OutputStream out = sender.getOutputStream();
for (ByteBuffer raw : raws) {
out.write(raw.array());
- ZMQ.msleep(100);
}
Msg msg = ZMQ.recv(receiver, 0);
assertThat(msg, notNullValue());
assertThat(new String(msg.data(), ZMQ.CHARSET), is(payload));
- ZMQ.msleep(500);
- assertThat(monitor.events.size(), is(1));
- assertThat(monitor.events.get(0).event, is(ZMQ.ZMQ_EVENT_HANDSHAKE_PROTOCOL));
- assertThat((Integer) monitor.events.get(0).arg, is(version));
+ monitor.join();
+
+ final Event event = monitor.events[0];
+ assertThat(event, notNullValue());
+ assertThat(event.event, is(ZMQ.ZMQ_EVENT_HANDSHAKE_PROTOCOL));
+ assertThat((Integer) event.arg, is(version));
InputStream in = sender.getInputStream();
byte[] data = new byte[255];
diff --git a/src/test/java/zmq/io/V0ProtocolTest.java b/src/test/java/zmq/io/V0ProtocolTest.java
index bd547d23..1a5b7aef 100644
--- a/src/test/java/zmq/io/V0ProtocolTest.java
+++ b/src/test/java/zmq/io/V0ProtocolTest.java
@@ -10,6 +10,18 @@
public class V0ProtocolTest extends AbstractProtocolVersion
{
+ @Test
+ public void testFixIssue524() throws IOException, InterruptedException
+ {
+ for (int idx = 0; idx < REPETITIONS; ++idx) {
+ if (idx % 100 == 0) {
+ System.out.print(idx + " ");
+ }
+ testProtocolVersion0short();
+ }
+ System.out.println();
+ }
+
@Test(timeout = 2000)
public void testProtocolVersion0short() throws IOException, InterruptedException
{
diff --git a/src/test/java/zmq/io/V1ProtocolTest.java b/src/test/java/zmq/io/V1ProtocolTest.java
index e1045f34..764159d0 100644
--- a/src/test/java/zmq/io/V1ProtocolTest.java
+++ b/src/test/java/zmq/io/V1ProtocolTest.java
@@ -10,7 +10,19 @@
public class V1ProtocolTest extends AbstractProtocolVersion
{
- @Test(timeout = 2000)
+ @Test
+ public void testFixIssue524() throws IOException, InterruptedException
+ {
+ for (int idx = 0; idx < REPETITIONS; ++idx) {
+ if (idx % 100 == 0) {
+ System.out.print(idx + " ");
+ }
+ testProtocolVersion1short();
+ }
+ System.out.println();
+ }
+
+ @Test
public void testProtocolVersion1short() throws IOException, InterruptedException
{
List<ByteBuffer> raws = raws(0);
@@ -25,7 +37,7 @@ public void testProtocolVersion1short() throws IOException, InterruptedException
assertProtocolVersion(1, raws, "abcdefg");
}
- @Test(timeout = 2000)
+ @Test
public void testProtocolVersion1long() throws IOException, InterruptedException
{
List<ByteBuffer> raws = raws(0);
diff --git a/src/test/java/zmq/io/V2ProtocolTest.java b/src/test/java/zmq/io/V2ProtocolTest.java
index d5e64bce..7fda31bc 100644
--- a/src/test/java/zmq/io/V2ProtocolTest.java
+++ b/src/test/java/zmq/io/V2ProtocolTest.java
@@ -21,7 +21,19 @@ protected ByteBuffer identity()
.put((byte) 0);
}
- @Test(timeout = 2000)
+ @Test
+ public void testFixIssue524() throws IOException, InterruptedException
+ {
+ for (int idx = 0; idx < REPETITIONS; ++idx) {
+ if (idx % 100 == 0) {
+ System.out.print(idx + " ");
+ }
+ testProtocolVersion2short();
+ }
+ System.out.println();
+ }
+
+ @Test
public void testProtocolVersion2short() throws IOException, InterruptedException
{
List<ByteBuffer> raws = raws(1);
@@ -38,7 +50,7 @@ public void testProtocolVersion2short() throws IOException, InterruptedException
assertProtocolVersion(2, raws, "abcdefg");
}
- @Test(timeout = 2000)
+ @Test
public void testProtocolVersion2long() throws IOException, InterruptedException
{
List<ByteBuffer> raws = raws(1);
|
