Skip to content

Commit cb3e082

Browse files
somdoronsomdoron
committed
problem: ZMTP v2Test is flapping
Fixed it by using raw sockets instead of stream socket
1 parent 8a7c497 commit cb3e082

File tree

1 file changed

+70
-67
lines changed

1 file changed

+70
-67
lines changed

‎src/NetMQ.Tests/ZMTPTests.cs

Lines changed: 70 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -23,52 +23,49 @@ public class ZMTPTests : IClassFixture<CleanupAfterFixture>
2323
{
2424
public ZMTPTests() => NetMQConfig.Cleanup();
2525

26+
private byte[] ReadRawXBytes(Socket raw, int toRead)
27+
{
28+
var bytes = new byte[toRead];
29+
int read = raw.Receive(bytes);
30+
31+
while (read < toRead)
32+
read += raw.Receive(bytes, read, toRead - read, SocketFlags.None);
33+
34+
return bytes;
35+
}
36+
2637
[Fact]
2738
public void V2Test()
2839
{
29-
using (var raw = new StreamSocket())
30-
using (var socket = new DealerSocket())
31-
{
32-
int port = raw.BindRandomPort("tcp://*");
33-
socket.Connect($"tcp://localhost:{port}");
40+
using var raw = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
41+
using var socket = new DealerSocket();
3442

35-
var routingId = raw.ReceiveFrameBytes();
36-
var preamble = raw.ReceiveFrameBytes();
37-
Assert.Equal(10, preamble.Length);
43+
int port = socket.BindRandomPort("tcp://*");
44+
raw.Connect("127.0.0.1", port);
3845

39-
raw.SendMoreFrame(routingId).SendFrame(new byte[] {0xff,0,0,0,0,0,0,0,0,0x7f,1,5}); // Signature
40-
raw.SendMoreFrame(routingId).SendFrame(new byte[] {0, 0}); // Empty Identity
41-
raw.SendMoreFrame(routingId).SendFrame(new byte[] {0, 1, 5}); // One byte message
42-
43-
// Receive rest of the greeting
44-
raw.SkipFrame(); // RoutingId
45-
var signature = raw.ReceiveFrameBytes();
46-
47-
if (signature.Length == 2)
48-
Assert.Equal(new byte[] {3, 5}, signature);
49-
else if (signature.Length == 1)
50-
{
51-
// Receive rest of the greeting
52-
raw.SkipFrame(); // RoutingId
53-
signature = raw.ReceiveFrameBytes();
54-
Assert.Equal(new byte[] {5}, signature);
55-
}
56-
57-
// Receive the identity
58-
raw.SkipFrame(); // RoutingId
59-
var identity = raw.ReceiveFrameBytes();
60-
Assert.Equal(new byte[] {0, 0}, identity);
61-
62-
// Receiving msg send by the raw
63-
var msg = socket.ReceiveFrameBytes();
64-
Assert.Equal(new byte[] {5}, msg);
65-
66-
// Sending msg from socket to raw
67-
socket.SendFrame(new byte[]{6});
68-
raw.SkipFrame(); // RoutingId
69-
msg = raw.ReceiveFrameBytes();
70-
Assert.Equal(new byte[]{0,1,6}, msg);
71-
}
46+
// preamble
47+
ReadRawXBytes(raw, 10);
48+
49+
raw.Send(new byte[] { 0xff, 0, 0, 0, 0, 0, 0, 0, 0, 0x7f, 1, 5 }); // Signature
50+
raw.Send(new byte[] { 0, 0 }); // Empty Identity
51+
raw.Send(new byte[] { 0, 1, 5 }); // One byte message
52+
53+
// Receive rest of the greeting
54+
var signature = ReadRawXBytes(raw, 2);
55+
Assert.Equal(new byte[] { 3, 5 }, signature);
56+
57+
// Receive the identity
58+
var identity = ReadRawXBytes(raw, 2);
59+
Assert.Equal(new byte[] { 0, 0 }, identity);
60+
61+
// Receiving msg send by the raw
62+
var msg = socket.ReceiveFrameBytes();
63+
Assert.Equal(new byte[] { 5 }, msg);
64+
65+
// Sending msg from socket to raw
66+
socket.SendFrame(new byte[] { 6 });
67+
msg = ReadRawXBytes(raw, 3);
68+
Assert.Equal(new byte[] { 0, 1, 6 }, msg);
7269
}
7370

7471
[Fact]
@@ -79,14 +76,14 @@ public void HeartbeatEnabled()
7976
{
8077
sub.Options.HeartbeatInterval = TimeSpan.FromMilliseconds(10);
8178
sub.Options.HeartbeatTimeout = TimeSpan.FromMilliseconds(1);
82-
79+
8380
int port = pub.BindRandomPort("tcp://*");
8481
sub.Connect($"tcp://localhost:{port}");
85-
82+
8683
Thread.Sleep(3000);
8784
}
8885
}
89-
86+
9087
[Fact]
9188
public void V3Test()
9289
{
@@ -104,56 +101,62 @@ public void V3Test()
104101
raw.SendMoreFrame(routingId).SendFrame(
105102
new byte[64]
106103
{
107-
0xff,0,0,0,0,0,0,0,0,0x7f,3,0, (byte)'N', (byte)'U', (byte)'L', (byte)'L',
108-
0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0
104+
0xff, 0, 0, 0, 0, 0, 0, 0, 0, 0x7f, 3, 0, (byte) 'N', (byte) 'U', (byte) 'L', (byte) 'L',
105+
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
106+
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
109107
}); // V3 Greeting
110108
raw.SendMoreFrame(routingId).SendFrame(
111109
new byte[43]
112110
{
113111
4, 41,
114-
5, (byte)'R', (byte)'E',(byte)'A',(byte)'D',(byte)'Y',
115-
11, (byte)'S',(byte)'o',(byte)'c',(byte)'k',(byte)'e',(byte)'t',(byte)'-',(byte)'T',(byte)'y',(byte)'p',(byte)'e',
116-
0,0,0,6,(byte)'D',(byte)'E',(byte)'A',(byte)'L',(byte)'E',(byte)'R',
117-
8,(byte)'I',(byte)'d',(byte)'e',(byte)'n',(byte)'t',(byte)'i',(byte)'t',(byte)'y',
118-
0,0,0,0
112+
5, (byte) 'R', (byte) 'E', (byte) 'A', (byte) 'D', (byte) 'Y',
113+
11, (byte) 'S', (byte) 'o', (byte) 'c', (byte) 'k', (byte) 'e', (byte) 't', (byte) '-',
114+
(byte) 'T', (byte) 'y', (byte) 'p', (byte) 'e',
115+
0, 0, 0, 6, (byte) 'D', (byte) 'E', (byte) 'A', (byte) 'L', (byte) 'E', (byte) 'R',
116+
8, (byte) 'I', (byte) 'd', (byte) 'e', (byte) 'n', (byte) 't', (byte) 'i', (byte) 't',
117+
(byte) 'y',
118+
0, 0, 0, 0
119119
}); // Ready Command
120-
120+
121121
int read = 0;
122122
while (read < 64 + 43 - 10)
123123
{
124124
raw.SkipFrame(); // RoutingId
125125
var bytes = raw.ReceiveFrameBytes();
126126
read += bytes.Length;
127127
}
128-
129-
raw.SendMoreFrame(routingId).SendFrame(new byte[] {0, 1, 5}); // One byte message
130-
128+
129+
raw.SendMoreFrame(routingId).SendFrame(new byte[] { 0, 1, 5 }); // One byte message
130+
131131
// Receiving msg send by the raw
132132
var msg = socket.ReceiveFrameBytes();
133-
Assert.Equal(new byte[] {5}, msg);
134-
133+
Assert.Equal(new byte[] { 5 }, msg);
134+
135135
// Sending msg from socket to raw
136-
socket.SendFrame(new byte[]{6});
136+
socket.SendFrame(new byte[] { 6 });
137137
raw.SkipFrame(); // RoutingId
138138
msg = raw.ReceiveFrameBytes();
139-
Assert.Equal(new byte[]{0,1,6}, msg);
140-
139+
Assert.Equal(new byte[] { 0, 1, 6 }, msg);
140+
141141
// Sending ping message
142-
raw.SendMoreFrame(routingId).SendFrame(new byte[11] {4, 9, 4, (byte)'P', (byte)'I', (byte)'N', (byte)'G', 0, 0,(byte)'H', (byte)'I'});
143-
142+
raw.SendMoreFrame(routingId).SendFrame(new byte[11]
143+
{ 4, 9, 4, (byte) 'P', (byte) 'I', (byte) 'N', (byte) 'G', 0, 0, (byte) 'H', (byte) 'I' });
144+
144145
// Receive pong
145146
raw.SkipFrame(); // RoutingId
146147
var ping = raw.ReceiveFrameBytes();
147-
Assert.Equal(new byte[9] {4,7,4,(byte)'P', (byte)'O', (byte)'N', (byte)'G', (byte)'H', (byte)'I'}, ping);
148-
148+
Assert.Equal(
149+
new byte[9] { 4, 7, 4, (byte) 'P', (byte) 'O', (byte) 'N', (byte) 'G', (byte) 'H', (byte) 'I' },
150+
ping);
151+
149152
// We should receive ping now
150153
raw.SkipFrame();
151154
ping = raw.ReceiveFrameBytes();
152-
Assert.Equal(new byte[9] {4,7,4,(byte)'P', (byte)'I', (byte)'N', (byte)'G', 0, 0}, ping);
155+
Assert.Equal(new byte[9] { 4, 7, 4, (byte) 'P', (byte) 'I', (byte) 'N', (byte) 'G', 0, 0 }, ping);
153156
}
154157
}
155158

156-
#if NET47
159+
#if NET47
157160
[Fact]
158161
public void WithLibzmq()
159162
{
@@ -174,6 +177,6 @@ public void WithLibzmq()
174177
}
175178
}
176179

177-
#endif
180+
#endif
178181
}
179182
}

0 commit comments

Comments
 (0)