Skip to content

Commit dceb8ea

Browse files
jonathanstoweJonathan Stowe
jonathanstowe
authored and
Jonathan Stowe
committed
Implement server features.
Specifically this provides an implementation of the SUBSCRIBE/UNSUBSCRIBE commands and a Connection object. This is sufficient to make a Stomp server.
1 parent bb67f6e commit dceb8ea

7 files changed

Lines changed: 221 additions & 20 deletions

File tree

‎example/receiver‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
#!/usr/bin/env raku
2+
13
use Stomp::Client;
24

35
sub MAIN(:$login = 'guest', :$password = 'guest', :$port = 61613, :$host = 'localhost', :$queue = 'stomptest') {

‎example/sender‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
#!/usr/bin/env raku
2+
13
use Stomp::Client;
24

35
sub MAIN(Str $message = 'Hello, World', :$login = 'guest', :$password = 'guest', :$port = 61613, :$host = 'localhost', :$queue = 'stomptest') {

‎example/server‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
#!/usr/bin/env raku
2+
3+
use Stomp::Server;
4+
5+
sub MAIN(:$port = 61613, :$host = 'localhost') {
6+
my $server = Stomp::Server.new(:$port, :$host);
7+
8+
my $all-messages = Supplier.new;
9+
react {
10+
whenever $server.listen -> $v {
11+
whenever $v.published-messages -> $m {
12+
$all-messages.emit: $m;
13+
}
14+
whenever $all-messages -> $m {
15+
if $m ~~ $v {
16+
$v.send: $m;
17+
}
18+
}
19+
}
20+
}
21+
}
22+
23+
# vim: ft=raku

‎lib/Stomp/Message.rakumod‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@ class Stomp::Message {
33
has %.headers;
44
has $.body = '';
55

6+
has $!uuid;
7+
68
method Str() {
79
qq:to/END/
810
$!command
@@ -11,4 +13,9 @@ class Stomp::Message {
1113
$!body\0
1214
END
1315
}
16+
17+
method uuid() returns Str {
18+
use UUID;
19+
$!uuid //= UUID.new.Str;
20+
}
1421
}

‎lib/Stomp/Server.rakumod‎

Lines changed: 104 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,116 @@
11
use Stomp::Parser;
22
use Stomp::MessageStream;
3+
need Stomp::Message;
34

4-
class Stomp::Server does Stomp::MessageStream[Stomp::Parser::ClientCommands] {
5+
class Stomp::Server {
56
has Str $.host is required;
67
has Int $.port is required;
78

9+
enum AckMode ( Auto => 'auto', Client => 'client', Individual => 'client-individual');
10+
11+
class Subscription {
12+
has Str $.id is required;
13+
has Str $.destination is required;
14+
has AckMode $.ack = Auto;
15+
16+
method ACCEPTS(Stomp::Message $mess) {
17+
$!destination eq $mess.headers<destination>;
18+
}
19+
}
20+
21+
class Connection does Stomp::MessageStream[Stomp::Parser::ClientCommands] {
22+
has $.conn;
23+
has Supply $!messages;
24+
25+
has Subscription @.subscriptions;
26+
has Lock::Async $!subscription-lock;
27+
28+
has Supplier $!sent-message-supplier;
29+
has Promise $.connected;
30+
31+
submethod TWEAK {
32+
$!messages = self!process-messages($!conn.Supply).share;
33+
34+
$!subscription-lock = Lock::Async.new;
35+
36+
$!connected = Promise.new;
37+
38+
my &quit = {
39+
when X::Stomp::MalformedMessage {
40+
await $!conn.print: Stomp::Message.new:
41+
command => 'ERROR',
42+
body => .message;
43+
}
44+
};
45+
46+
my $connect-tap = $!messages.grep({ $_.command ~~ 'CONNECT'|'STOMP' }).tap:
47+
{
48+
await $!conn.print: Stomp::Message.new:
49+
command => 'CONNECTED',
50+
headers => ( version => '1.2' );
51+
$!connected.keep: True;
52+
$connect-tap.close;
53+
}, :&quit;
54+
$!messages.grep({ $_.command ~~ 'SUBSCRIBE' }).tap: {
55+
$!subscription-lock.protect: {
56+
@!subscriptions.push: Subscription.new( id => $_.headers<id>,
57+
destination => $_.headers<destination>,
58+
ack => $_.headers<ack> ?? AckMode($_.headers<ack>) !! Auto );
59+
};
60+
}, :&quit;
61+
$!messages.grep({ $_.command ~~ 'UNSUBSCRIBE' }).tap: {
62+
$!subscription-lock.protect: {
63+
my $id = $_.headers<id>;
64+
@!subscriptions = @!subscriptions.grep({ $_.id !~~ $id });
65+
};
66+
}, :&quit;
67+
68+
$!sent-message-supplier = Supplier.new;
69+
$!messages.grep({ $_.command ~~ 'SEND' }).tap: {
70+
$!sent-message-supplier.emit: $_;
71+
}, :&quit;
72+
}
73+
74+
method published-messages() returns Supply {
75+
$!sent-message-supplier.Supply;
76+
}
77+
78+
method ACCEPTS(Stomp::Message $mess) {
79+
$!subscription-lock.protect({
80+
@!subscriptions.map(*.destination).any eq $mess.headers<destination>;
81+
});
82+
}
83+
84+
method subscription-for-message(Stomp::Message $mess) {
85+
$!subscription-lock.protect({
86+
@!subscriptions.first({ $_.destination eq $mess.headers<destination>});
87+
});
88+
}
89+
90+
method send(Stomp::Message $mess) {
91+
if self.subscription-for-message($mess) -> $sub {
92+
my $subscription = $sub.id;
93+
my $message-id = $mess.uuid;
94+
my $out = Stomp::Message.new(
95+
command => 'MESSAGE',
96+
body => $mess.body,
97+
headers => (|$mess.headers, :$message-id, :$subscription, )
98+
);
99+
$!conn.print: $out;
100+
}
101+
else {
102+
fail "Not subscribed to message";
103+
}
104+
}
105+
}
106+
8107
method listen() {
9108
supply {
10109
whenever self.socket-provider.listen($!host, $!port) -> $conn {
11-
self!process-messages($conn).tap:
12-
{
13-
await $conn.print: Stomp::Message.new:
14-
command => 'CONNECTED',
15-
headers => ( accept-version => '1.2' );
16-
},
17-
quit => {
18-
when X::Stomp::MalformedMessage {
19-
await $conn.print: Stomp::Message.new:
20-
command => 'ERROR',
21-
body => .message;
22-
}
23-
};
110+
my $connection = Connection.new(:$conn);
111+
whenever $connection.connected {
112+
emit $connection;
113+
}
24114
}
25115
}
26116
}

‎t/001-meta.rakutest‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
#!raku
2+
13
use Test;
24

35
my Bool $got-test-meta = True;

‎t/server.rakutest‎

Lines changed: 81 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ use Test;
22
use Test::IO::Socket::Async;
33
use Stomp;
44

5-
plan 15;
5+
plan 28;
66

77
constant $test-socket = Test::IO::Socket::Async.new;
88
my \TestableServer = Stomp::Server but role {
@@ -38,25 +38,100 @@ dies-ok { TestableServer.new(port => $test-port) }, "Must provide host and port
3838
constant $test-password = 'correcthorsebatterystaple';
3939

4040
my $test-server = TestableServer.new(host => $test-host, port => $test-port);
41-
my $listen-tap = $test-server.listen().tap(-> $conn { });
41+
my $client-connection;
42+
my $listen-tap = $test-server.listen().tap(-> $conn { $client-connection = $conn });
4243
my $socket-listener = await $test-socket.start-listening;
4344
my $test-conn = $socket-listener.incoming-connection;
4445

4546
$test-conn.receive-data: Stomp::Message.new(
4647
command => 'CONNECT',
4748
headers => (
48-
login => $test-login,
49-
passcode => $test-password,
50-
accept-version => '1.2'
49+
host => 'localhost',
50+
login => $test-login,
51+
passcode => $test-password,
52+
accept-version => '1.2'
5153
));
5254

5355
my $message-text = await $test-conn.sent-data;
5456
my $parsed-message = Stomp::Parser.parse($message-text);
5557
ok $parsed-message, "Server responded to CONNECT with valid message";
5658
my $message = $parsed-message.made;
5759
is $message.command, "CONNECTED", "Server sent CONNECTED command";
58-
ok $message.headers<accept-version>:exists, "Server sent accept-version header";
60+
ok $message.headers<version>:exists, "Server sent version header";
5961
is $message.body, "", "Server sent no message body";
62+
isa-ok $client-connection, 'Stomp::Server::Connection', "and the tap received the correct object";
63+
}
64+
65+
{
66+
constant $test-login = 'user';
67+
constant $test-password = 'correcthorsebatterystaple';
68+
69+
my $test-server = TestableServer.new(host => $test-host, port => $test-port);
70+
my $client-connection;
71+
my $listen-tap = $test-server.listen().tap(-> $conn { $client-connection = $conn });
72+
my $socket-listener = await $test-socket.start-listening;
73+
my $test-conn = $socket-listener.incoming-connection;
74+
75+
$test-conn.receive-data: Stomp::Message.new(
76+
command => 'CONNECT',
77+
headers => (
78+
login => $test-login,
79+
passcode => $test-password,
80+
accept-version => '1.2'
81+
));
82+
83+
my $message-text = await $test-conn.sent-data;
84+
85+
is $client-connection.subscriptions.elems, 0, "new connection doesn't have a subscription";
86+
87+
my $destination = '/queue/testqueue';
88+
my $id = 1;
89+
$test-conn.receive-data: Stomp::Message.new(
90+
command => 'SUBSCRIBE',
91+
headers => (
92+
:$destination,
93+
:$id,
94+
));
95+
96+
my $match-message = Stomp::Message.new(
97+
command => 'SEND',
98+
headers => (
99+
:$destination
100+
),
101+
body => 'Test Message'
102+
);
103+
104+
is $client-connection.subscriptions.elems, 1, "now have one subscription";
105+
is $client-connection.subscriptions.first.id, $id, "and it has the right id";
106+
is $client-connection.subscriptions.first.destination, $destination, "and it has the right destination";
107+
is $client-connection.subscriptions.first.ack, 'auto', "and the ack is 'auto'";
108+
ok $match-message ~~ $client-connection.subscriptions.first , "subscription matches message to that destination";
109+
ok $client-connection.subscription-for-message($match-message), "subscription-for-message";
110+
ok $match-message ~~ $client-connection, "connection matches message to that destination";
111+
112+
my $rec-message;
113+
my $rec-promise = Promise.new;
114+
$client-connection.published-messages.tap({
115+
$rec-message = $_;
116+
$rec-promise.keep: True;
117+
});
118+
119+
$test-conn.receive-data: $match-message;
120+
121+
await Promise.anyof($rec-promise, Promise.in(5));
122+
123+
is $rec-message.body, $match-message.body, "got the message from published-messages";
124+
125+
$test-conn.receive-data: Stomp::Message.new(
126+
command => 'UNSUBSCRIBE',
127+
headers => (
128+
:$destination,
129+
:$id,
130+
));
131+
is $client-connection.subscriptions.elems, 0, "now have no subscription after unsubscribe";
132+
ok $match-message !~~ $client-connection.subscriptions.first , "subscription no longer matches message to that destination";
133+
ok $match-message !~~ $client-connection, "connection no longer matches message to that destination";
134+
60135
}
61136

62137
{

0 commit comments

Comments
 (0)