Class: Slack::RealTime::Concurrency::Celluloid::Socket

Inherits:
Socket
  • Object
show all
Extended by:
Forwardable
Includes:
Celluloid::IO, Celluloid::Internals::Logger
Defined in:
lib/slack/real_time/concurrency/celluloid.rb

Defined Under Namespace

Classes: Actor

Constant Summary collapse

BLOCK_SIZE =
4096

Instance Attribute Summary collapse

Attributes inherited from Socket

#driver, #options, #url

Instance Method Summary collapse

Methods inherited from Socket

#current_time, #send_data, #start_sync, #time_since_last_message

Constructor Details

#initialize(*args) ⇒ Socket


22
23
24
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 22

def initialize(*args)
  super
end

Instance Attribute Details

#socketObject (readonly)

Returns the value of attribute socket


20
21
22
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 20

def socket
  @socket
end

Instance Method Details

#closeObject


47
48
49
50
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 47

def close
  @closing = true
  super
end

#connect!Object


26
27
28
29
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 26

def connect!
  super
  run_loop
end

#connected?Boolean


97
98
99
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 97

def connected?
  !@connected.nil? && !@driver.nil?
end

#disconnect!Object


42
43
44
45
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 42

def disconnect!
  super
  @ping_timer.cancel if @ping_timer
end

#handle_read(buffer) ⇒ Object


59
60
61
62
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 59

def handle_read(buffer)
  logger.debug("#{self.class}##{__method__}") { buffer }
  driver.parse buffer if driver
end

#readObject

Raises:

  • (EOFError)

52
53
54
55
56
57
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 52

def read
  buffer = socket.readpartial(BLOCK_SIZE)
  raise EOFError unless buffer && !buffer.empty?

  async.handle_read(buffer)
end

#restart_async(client, new_url) ⇒ Object


90
91
92
93
94
95
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 90

def restart_async(client, new_url)
  @last_message_at = current_time
  @url = new_url
  @client = client
  Actor.new(future.run_client_loop)
end

#run_client_loopObject


75
76
77
78
79
80
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 75

def run_client_loop
  @client.run_loop
rescue StandardError => e
  logger.debug("#{self.class}##{__method__}") { e }
  raise e
end

#run_loopObject


31
32
33
34
35
36
37
38
39
40
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 31

def run_loop
  @closing = false
  @socket = build_socket
  @connected = @socket.connect
  driver.start
  loop { read } if socket
rescue EOFError, Errno::ECONNRESET, Errno::EPIPE => e
  logger.debug("#{self.class}##{__method__}") { e }
  driver.emit(:close, WebSocket::Driver::CloseEvent.new(1001, 'server closed connection')) unless @closing
end

#run_ping_loopObject


82
83
84
85
86
87
88
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 82

def run_ping_loop
  return unless @client.run_ping?

  @ping_timer = every @client.websocket_ping do
    @client.run_ping!
  end
end

#start_async(client) ⇒ Object


69
70
71
72
73
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 69

def start_async(client)
  @client = client
  Actor.new(future.run_ping_loop)
  Actor.new(future.run_client_loop)
end

#write(data) ⇒ Object


64
65
66
67
# File 'lib/slack/real_time/concurrency/celluloid.rb', line 64

def write(data)
  logger.debug("#{self.class}##{__method__}") { data }
  socket.write(data)
end