Class: CableRoom::RoomMembership

Inherits:
Object
  • Object
show all
Includes:
Ports
Defined in:
lib/cable_room/room_member.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Ports

#close_streamed_ports!, #ports

Constructor Details

#initialize(cable_channel, room_class, room_key, create: false, on_room_opened: nil, on_joined: nil, on_message: nil, on_room_closed: nil, on_left: nil, tags: [], extra: nil, &preconfigure) ⇒ RoomMembership

Returns a new instance of RoomMembership.



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
 
# File 'lib/cable_room/room_member.rb', line 51

def initialize(
  cable_channel,
  room_class,
  room_key,
  create: false,
  on_room_opened: nil, # Not guaranteed (Only given if the Room Opens while we are trying to connect to it)
  on_joined: nil,
  on_message: nil,
  on_room_closed: nil, # Not guaranteed (Only given if the Room Closes while we are connected to it)
  on_left: nil,
  tags: [],
  extra: nil,
  &preconfigure
)
  @mutex = Monitor.new

  @has_left = false
  @has_established = false

  @cable_channel = cable_channel
  @room_class = room_class
  @room_key = room_key
  @allow_create = create

  @on_room_opened = on_room_opened
  @on_joined = on_joined
  @on_message = on_message
  @on_room_closed = on_room_closed
  @on_left = on_left
  @preconfigure = preconfigure

  @tags = Array(tags).map(&:to_sym)
  @extra = extra

  initiate_connection
end

Instance Attribute Details

#room_class ⇒ Object (readonly)

Returns the value of attribute room_class.



49
50
51
 
# File 'lib/cable_room/room_member.rb', line 49

def room_class
  @room_class
end

Instance Method Details

#<<(data) ⇒ Object 



90
91
92
 
# File 'lib/cable_room/room_member.rb', line 90

def <<(data)
  port_transmit(room_class::ROOM_IN_CHANNEL, data)
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


94
95
96
 
# File 'lib/cable_room/room_member.rb', line 94

def connected?
  !left? && @has_established
end

#key ⇒ Object 



141
 
# File 'lib/cable_room/room_member.rb', line 141

def key; @room_key; end

#leave! ⇒ Object 



120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/cable_room/room_member.rb', line 120

def leave!
  @mutex.synchronize do
    return if left?

    close_streamed_ports!
    @cable_channel._room_memberships.delete(self)
    port_transmit(room_class::ROOM_IN_CHANNEL, { type: 'port_disconnected' }, secure_context: true)
    @has_left = true
  end
  @on_left&.call(self)
end

#left? ⇒ Boolean

Returns:

  • (Boolean) 


98
99
100
 
# File 'lib/cable_room/room_member.rb', line 98

def left?
  @mutex.synchronize { @has_left }
end

#ping! ⇒ Object 



102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/cable_room/room_member.rb', line 102

def ping!
  return if left?
  if @has_established
    port_transmit(room_class::ROOM_IN_CHANNEL, { type: 'port_ping' }, secure_context: true)
  elsif @streams_live
    # Announced but never acknowledged. Our streams were already live when we announced (see
    # initiate_connection), so this is not the subscribe race — the announcement or the
    # acknowledgement was lost in transit, or the room did not exist yet. An unestablished
    # membership silently drops everything the client sends (see #<<), so re-announce instead
    # of pinging: port_connected is idempotent on the room side (the port is merged,
    # user_joined fires only once).
    transmit_port_connected
  end
  @mutex.synchronize do
    maybe_provision_room
  end
end

#rejoin! ⇒ Object 



132
133
134
135
136
137
138
139
 
# File 'lib/cable_room/room_member.rb', line 132

def rejoin!
  @mutex.synchronize do
    return unless left?

    @has_left = false
    initiate_connection
  end
end

#stream_port(port, on_live: nil, **kwargs, &blk) ⇒ Object

Every stream this membership opens reports back once it is live; the last one to come up sends port_connected (see initiate_connection).

The adapter runs the liveness callback on its own terms — the async and inline adapters call it while holding their subscriber-map lock, the Redis adapter on the event loop — so nothing may happen inside it except a hop to a worker thread: taking @mutex there deadlocks against an initiate_connection that is mid-broadcast, and broadcasting from it re-enters the lock.



150
151
152
153
154
155
156
157
158
 
# File 'lib/cable_room/room_member.rb', line 150

def stream_port(port, on_live: nil, **kwargs, &blk)
  generation = @stream_generation
  @streams_pending += 1
  connection = @cable_channel.connection
  super(port, **kwargs, on_live: lambda do
    connection.worker_pool.async_invoke(self, :_stream_became_live, generation, connection: connection)
    on_live&.call
  end, &blk)
end