Skip to content

Commit f68b822

Browse files
rmarronnierrmarronnier
authored
Cable::Server#active_connections_for and Cable::Server#subscribed_channels_for public checkup methods (#86)
* New methods + using existing helper * DRY suggestions * Minimal specs * Typo * connection#close fix for channel-less connections * server#active_connections_for specs * Warning comments for methods * last server specs + helpers * More explicit method return type * Ameba correction
1 parent f0e3f9f commit f68b822

7 files changed

Lines changed: 141 additions & 70 deletions

File tree

‎spec/cable/connection_spec.cr‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,22 @@ require "../spec_helper"
33
include RequestHelpers
44

55
describe Cable::Connection do
6-
it "removes the connection channel on close" do
7-
connect do |connection, _socket|
8-
connection.receive({"command" => "subscribe", "identifier" => {channel: "ChatChannel", room: "1"}.to_json}.to_json)
9-
ConnectionTest::CHANNELS.keys.size.should eq(1)
6+
describe "#close" do
7+
it "closes the connection socket even without channel subscriptions" do
8+
connect do |connection, _socket|
9+
connection.closed?.should eq(false)
10+
connection.close
11+
connection.closed?.should eq(true)
12+
end
13+
end
14+
it "removes the connection channel on close" do
15+
connect do |connection, _socket|
16+
connection.receive({"command" => "subscribe", "identifier" => {channel: "ChatChannel", room: "1"}.to_json}.to_json)
17+
ConnectionTest::CHANNELS.keys.size.should eq(1)
18+
connection.close
19+
ConnectionTest::CHANNELS.keys.size.should eq(0)
20+
end
1021
end
11-
ConnectionTest::CHANNELS.keys.size.should eq(0)
1222
end
1323

1424
describe "#receive" do

‎spec/cable/handler_spec.cr‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ describe Cable::Handler do
5050
ws2 = HTTP::WebSocket.new("ws://#{listen_address}/updates?test_token=1")
5151

5252
Cable.server.connections.size.should eq(1)
53+
Cable.server.active_connections_for("1").size.should eq(1)
54+
Cable.server.subscribed_channels_for("1").size.should eq(0)
5355

5456
messages = [
5557
{type: "welcome"}.to_json,
@@ -69,6 +71,8 @@ describe Cable::Handler do
6971
ws2.run
7072

7173
Cable.server.connections.size.should eq(1)
74+
Cable.server.active_connections_for("1").size.should eq(1)
75+
Cable.server.subscribed_channels_for("1").size.should eq(1)
7276
end
7377

7478
it "malformed data from client" do

‎spec/cable/server_spec.cr‎

Lines changed: 80 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,7 @@ describe Cable::Server do
77
it "finds the connection and disconnects it" do
88
Cable.reset_server
99
Cable.temp_config(backend_class: Cable::DevBackend) do
10-
socket = DummySocket.new(IO::Memory.new)
11-
request = builds_request("abc123")
12-
connection = ApplicationCable::Connection.new(request, socket)
10+
connection = creates_new_connection("abc123")
1311
Cable.server.add_connection(connection)
1412
connection.connection_identifier.should contain("abc123")
1513

@@ -19,4 +17,83 @@ describe Cable::Server do
1917
end
2018
end
2119
end
20+
21+
describe "#active_connections_for" do
22+
it "accurately returns active connections for a specificic token" do
23+
Cable.reset_server
24+
Cable.temp_config(backend_class: Cable::DevBackend) do
25+
Cable.server.active_connections_for("abc123").size.should eq(0)
26+
Cable.server.active_connections_for("def456").size.should eq(0)
27+
28+
connection = creates_new_connection("abc123")
29+
Cable.server.add_connection(connection)
30+
31+
Cable.server.active_connections_for("abc123").size.should eq(1)
32+
33+
other_connection = creates_new_connection("def456")
34+
Cable.server.add_connection(other_connection)
35+
36+
Cable.server.active_connections_for("def456").size.should eq(1)
37+
38+
connection.close
39+
40+
Cable.server.active_connections_for("abc123").size.should eq(0)
41+
Cable.server.active_connections_for("def456").size.should eq(1)
42+
43+
other_connection.close
44+
45+
Cable.server.active_connections_for("def456").size.should eq(0)
46+
end
47+
end
48+
end
49+
50+
describe "#subscribed_channels_for" do
51+
it "accurately returns active channel subscriptions for a specificic token" do
52+
Cable.reset_server
53+
Cable.temp_config(backend_class: Cable::DevBackend) do
54+
connection_1 = creates_new_connection("aa")
55+
connection_2 = creates_new_connection("bb")
56+
57+
Cable.server.add_connection(connection_1)
58+
Cable.server.add_connection(connection_2)
59+
60+
Cable.server.subscribed_channels_for("aa").size.should eq(0)
61+
Cable.server.subscribed_channels_for("bb").size.should eq(0)
62+
63+
connection_1.subscribe(subscribe_payload("room_a"))
64+
65+
Cable.server.subscribed_channels_for("aa").size.should eq(1)
66+
Cable.server.subscribed_channels_for("bb").size.should eq(0)
67+
68+
connection_1.subscribe(subscribe_payload("room_b"))
69+
70+
Cable.server.subscribed_channels_for("aa").size.should eq(2)
71+
Cable.server.subscribed_channels_for("bb").size.should eq(0)
72+
73+
connection_2.subscribe(subscribe_payload("room_a"))
74+
75+
Cable.server.subscribed_channels_for("aa").size.should eq(2)
76+
Cable.server.subscribed_channels_for("bb").size.should eq(1)
77+
78+
connection_1.close
79+
connection_2.close
80+
end
81+
Cable.reset_server
82+
end
83+
end
84+
end
85+
86+
def creates_new_connection(token : String | Nil) : ApplicationCable::Connection
87+
ApplicationCable::Connection.new(builds_request(token: token), DummySocket.new(IO::Memory.new))
88+
end
89+
90+
def subscribe_payload(room : String) : Cable::Payload
91+
payload_json = {
92+
command: "subscribe",
93+
identifier: {
94+
channel: "ChatChannel",
95+
room: room,
96+
}.to_json,
97+
}.to_json
98+
Cable::Payload.from_json(payload_json)
2299
end

‎src/cable/channel.cr‎

Lines changed: 5 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ module Cable
5959
@stream_identifier = stream_identifier.to_s
6060
end
6161

62-
def self.broadcast_to(channel : String, message : JSON::Any)
62+
def self.broadcast_to(channel : String, message : JSON::Any | Hash(String, String))
6363
Cable::Logger.info { "[ActionCable] Broadcasting to #{channel}: #{message}" }
6464
Cable.server.publish(channel, message.to_json)
6565
end
@@ -70,60 +70,19 @@ module Cable
7070
Cable.server.publish(channel, message)
7171
end
7272

73-
def self.broadcast_to(channel : String, message : Hash(String, String))
74-
Cable::Logger.info { "[ActionCable] Broadcasting to #{channel}: #{message}" }
75-
Cable.server.publish(channel, message.to_json)
76-
end
77-
78-
def broadcast(message : String)
79-
if stream_id = stream_identifier.presence
80-
Cable::Logger.info { "[ActionCable] Broadcasting to #{self.class}: #{message}" }
81-
Cable.server.send_to_channels(stream_id, message)
82-
else
83-
Cable::Logger.error { "#{self.class}.transmit(message : String) with #{message} without already using stream_from(stream_identifier)" }
84-
end
85-
end
86-
87-
def broadcast(message : JSON::Any)
73+
def broadcast(message : String | JSON::Any | Hash(String, String))
8874
if stream_id = stream_identifier.presence
8975
Cable::Logger.info { "[ActionCable] Broadcasting to #{self.class}: #{message}" }
9076
Cable.server.send_to_channels(stream_id, message)
9177
else
92-
Cable::Logger.error { "#{self.class}.transmit(message : JSON::Any) with #{message} without already using stream_from(stream_identifier)" }
78+
Cable::Logger.error { "#{self.class}.transmit(message : #{message.class}) with #{message} without already using stream_from(stream_identifier)" }
9379
end
9480
end
9581

96-
def broadcast(message : Hash(String, String))
97-
if stream_id = stream_identifier.presence
98-
Cable::Logger.info { "[ActionCable] Broadcasting to #{self.class}: #{message}" }
99-
Cable.server.send_to_channels(stream_id, message.to_json)
100-
else
101-
Cable::Logger.error { "#{self.class}.transmit(message : Hash(String, String)) with #{message} without already using stream_from(stream_identifier)" }
102-
end
103-
end
104-
105-
# broadcast single message to single connection for this channel
106-
def transmit(message : String)
107-
Cable::Logger.info { "[ActionCable] transmitting to #{self.class}: #{message}" }
108-
connection.socket.send({
109-
identifier: identifier,
110-
message: Cable.server.safe_decode_message(message),
111-
}.to_json)
112-
end
113-
114-
# broadcast single message to single connection for this channel
115-
def transmit(message : JSON::Any)
116-
Cable::Logger.info { "[ActionCable] transmitting to #{self.class}: #{message}" }
117-
connection.socket.send({
118-
identifier: identifier,
119-
message: Cable.server.safe_decode_message(message),
120-
}.to_json)
121-
end
122-
12382
# broadcast single message to single connection for this channel
124-
def transmit(message : Hash(String, String))
83+
def transmit(message : String | JSON::Any | Hash(String, String))
12584
Cable::Logger.info { "[ActionCable] transmitting to #{self.class}: #{message}" }
126-
connection.socket.send({
85+
connection.send_message({
12786
identifier: identifier,
12887
message: Cable.server.safe_decode_message(message),
12988
}.to_json)

‎src/cable/connection.cr‎

Lines changed: 21 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ require "uuid"
22

33
module Cable
44
abstract class Connection
5-
class UnathorizedConnectionException < Exception; end
5+
class UnauthorizedConnectionException < Exception; end
66

77
property internal_identifier : String = "0"
88
property connection_identifier : String = ""
@@ -38,7 +38,7 @@ module Cable
3838
# gather connection_identifier after the connection has gathered the id from identified_by(field)
3939
self.connection_identifier = "#{internal_identifier}-#{UUID.random}"
4040
subscribe_to_internal_channel
41-
rescue e : UnathorizedConnectionException
41+
rescue e : UnauthorizedConnectionException
4242
reject_connection!
4343
unsubscribe_from_internal_channel
4444
socket.close(HTTP::WebSocket::CloseCode::NormalClosure, "Farewell")
@@ -52,36 +52,43 @@ module Cable
5252
@connection_rejected = true
5353
end
5454

55+
def channels : Array(Cable::Channel)
56+
return Array(Cable::Channel).new unless Connection::CHANNELS.has_key?(connection_identifier)
57+
Connection::CHANNELS.[connection_identifier].values
58+
end
59+
5560
def closed? : Bool
5661
socket.closed?
5762
end
5863

5964
def close
60-
return true unless Connection::CHANNELS.has_key?(connection_identifier)
65+
if Connection::CHANNELS.has_key?(connection_identifier)
66+
Connection::CHANNELS[connection_identifier].each do |identifier, channel|
67+
# the ordering here is important
68+
Connection::CHANNELS[connection_identifier].delete(identifier)
69+
channel.close
70+
rescue e : IO::Error
71+
Cable.settings.on_error.call(e, "IO::Error: #{e.message} -> #{self.class.name}#close")
72+
end
6173

62-
Connection::CHANNELS[connection_identifier].each do |identifier, channel|
63-
# the ordering here is important
64-
Connection::CHANNELS[connection_identifier].delete(identifier)
65-
channel.close
66-
rescue e : IO::Error
67-
Cable.settings.on_error.call(e, "IO::Error: #{e.message} -> #{self.class.name}#close")
74+
Connection::CHANNELS.delete(connection_identifier)
75+
unsubscribe_from_internal_channel
6876
end
6977

70-
Connection::CHANNELS.delete(connection_identifier)
71-
unsubscribe_from_internal_channel
72-
Cable::Logger.info { "Terminating connection #{connection_identifier}" }
78+
return true if closed?
7379

80+
Cable::Logger.info { "Terminating connection #{connection_identifier}" }
7481
socket.close
7582
end
7683

7784
def send_message(message : String)
78-
return if socket.closed?
85+
return if closed?
7986

8087
socket.send(message)
8188
end
8289

8390
def reject_unauthorized_connection
84-
raise UnathorizedConnectionException.new
91+
raise UnauthorizedConnectionException.new
8592
end
8693

8794
# Convert the `message` to a proper `Payload`.

‎src/cable/handler.cr‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ module Cable
4949
socket.close(HTTP::WebSocket::CloseCode::InvalidFramePayloadData, "Invalid message")
5050
Cable.server.remove_connection(connection_id)
5151
Cable.settings.on_error.call(e, "Cable::Handler#socket.on_message")
52-
rescue e : Cable::Connection::UnathorizedConnectionException
52+
rescue e : Cable::Connection::UnauthorizedConnectionException
5353
# handle unauthorized connections
5454
# no need to log them
5555
ws_pinger.stop

‎src/cable/server.cr‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,20 @@ module Cable
7070
connections.delete(connection_id).try(&.close)
7171
end
7272

73+
# You shouldn't rely on these following two methods
74+
# for an exhaustive array of connections and channels
75+
# if your application can spawn more than one Cable.server instance.
76+
77+
# Only returns connections opened on this instance.
78+
def active_connections_for(token : String) : Array(Connection)
79+
connections.values.select { |connection| connection.token == token && !connection.closed? }
80+
end
81+
82+
# Only returns channel subscriptions opened on this instance.
83+
def subscribed_channels_for(token : String) : Array(Channel)
84+
active_connections_for(token).sum(&.channels)
85+
end
86+
7387
def subscribe_channel(channel : Channel, identifier : String)
7488
@channel_mutex.synchronize do
7589
if !@channels.has_key?(identifier)
@@ -112,7 +126,7 @@ module Cable
112126
@channels[channel_identifier].each do |channel|
113127
# TODO: would be nice to have a test where we open two connections
114128
# close one, and make sure the other one receives the message
115-
if channel.connection.socket.closed?
129+
if channel.connection.closed?
116130
channel.close
117131
else
118132
Cable::Logger.info { "#{channel.class} transmitting #{parsed_message} (via streamed from #{channel.stream_identifier})" }

0 commit comments

Comments
 (0)