Class: Wunderbar::Channel

Inherits:
BuilderBase show all
Defined in:
lib/wunderbar/websocket.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from BuilderBase

#get_binding, #set_variables_from_params

Constructor Details

#initialize(port, limit, locals = nil) ⇒ Channel

Returns a new instance of Channel.



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/wunderbar/websocket.rb', line 12

def initialize(port, limit, locals=nil)
  # verify that the port is available
  TCPServer.new('0.0.0.0', port).close 

  super()
  @port = port
  @connected = @complete = false
  @onopen = @onmessage = @onerror = @onclose = Proc.new {}
  @_scope = Struct.new(:params).new({})
  @channel1 = EM::Channel.new
  @channel2 = EM::Channel.new
  @memory = []
  @memory_channel = @channel1.subscribe do |msg| 
    @memory << msg.chomp unless Symbol === msg
    @memory.shift while @connected and limit and @memory.length > limit
  end

  if locals
    @_scope = locals['_scope'] || @_scope
    set_variables_from_params(locals)
    _ :type => 'stdout', :line => locals['_scope'].methods.inspect
  end

  websocket.run
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(method, *args, &block) ⇒ Object 



114
115
116
117
118
119
120
 
# File 'lib/wunderbar/websocket.rb', line 114

def method_missing(method, *args, &block)
  if @_scope and @_scope.respond_to? :method
    @_scope.__send__ method, *args, &block
  else
    super
  end
end

Instance Attribute Details

#complete ⇒ Object

Returns the value of attribute complete.



10
11
12
 
# File 'lib/wunderbar/websocket.rb', line 10

def complete
  @complete
end

#connected ⇒ Object (readonly)

Returns the value of attribute connected.



10
11
12
 
# File 'lib/wunderbar/websocket.rb', line 10

def connected
  @connected
end

#port ⇒ Object (readonly)

Returns the value of attribute port.



10
11
12
 
# File 'lib/wunderbar/websocket.rb', line 10

def port
  @port
end

Instance Method Details

#<<(value) ⇒ Object 



138
139
140
 
# File 'lib/wunderbar/websocket.rb', line 138

def <<(value)
  @channel1.push(value.to_json)
end

#_(*args, &block) ⇒ Object 



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/wunderbar/websocket.rb', line 122

def _(*args, &block)
  if block or args.length > 1 
    begin
      builder = Wunderbar::JsonBuilder.new(Struct.new(:params).new({}))
      builder._! self
      builder._(*args, &block)
    rescue Exception => e
      self << {:type=>'stderr', :line=>e.inspect}
    end
  elsif args.length == 1
    @channel1.push(args.first.to_json)
  else
    self
  end
end

#close ⇒ Object 



164
165
166
167
168
 
# File 'lib/wunderbar/websocket.rb', line 164

def close
  @channel1.unsubscribe @memory_channel if @memory_channel
  EM::WebSocket.stop
  websocket.join    
end

#onclose(&block) ⇒ Object 



110
111
112
 
# File 'lib/wunderbar/websocket.rb', line 110

def onclose(&block)
  @onclose = block
end

#onerror(&block) ⇒ Object 



106
107
108
 
# File 'lib/wunderbar/websocket.rb', line 106

def onerror(&block)
  @onerror = block
end

#onmessage(&block) ⇒ Object 



102
103
104
 
# File 'lib/wunderbar/websocket.rb', line 102

def onmessage(&block)
  @onmessage = block
end

#onopen(&block) ⇒ Object 



98
99
100
 
# File 'lib/wunderbar/websocket.rb', line 98

def onopen(&block)
  @onopen = block
end

#pop(*args) ⇒ Object 



90
91
92
 
# File 'lib/wunderbar/websocket.rb', line 90

def pop(*args)
  @channel2.pop(*args)
end

#push(*args) ⇒ Object 



82
83
84
 
# File 'lib/wunderbar/websocket.rb', line 82

def push(*args)
  @channel1.push(*args)
end

#recv(*args) ⇒ Object 



94
95
96
 
# File 'lib/wunderbar/websocket.rb', line 94

def recv(*args)
  @channel2.pop(*args)
end

#send(*args) ⇒ Object 



86
87
88
 
# File 'lib/wunderbar/websocket.rb', line 86

def send(*args)
  @channel1.push(*args)
end

#subscribe(*args, &block) ⇒ Object 



74
75
76
 
# File 'lib/wunderbar/websocket.rb', line 74

def subscribe(*args, &block)
  @channel2.subscribe(*args, &block)
end

#system(command) ⇒ Object 



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/wunderbar/websocket.rb', line 142

def system(command)
  Open3.popen3(command) do |pin, pout, perr|
    _ :type=>:stdin, :line=>command
    [
      Thread.new do
        pout.sync=true
        _ :type=>:stdout, :line=>pout.readline.chomp until pout.eof?
      end,
      Thread.new do
        perr.sync=true
        _ :type=>:stderr, :line=>perr.readline.chomp until perr.eof?
      end,
      Thread.new { pin.close }
    ].each {|thread| thread.join}
  end
end

#unsubscribe(*args, &block) ⇒ Object 



78
79
80
 
# File 'lib/wunderbar/websocket.rb', line 78

def unsubscribe(*args, &block)
  @channel2.unsubscribe(*args, &block)
end

#websocket ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/wunderbar/websocket.rb', line 38

def websocket
  return @websocket if @websocket
  ready = false
  @websocket = Thread.new do
    EM.epoll
    EM.run do
      connection = EventMachine::WebSocket::Connection
      EM.start_server('0.0.0.0', @port, connection, {}) do |ws|
        ws.onopen do
          @onopen.call(ws)
          @memory.each {|msg| ws.send msg }
          @connected = true
          ws.close_websocket if complete
        end
    
        sid = @channel1.subscribe do |msg| 
          if msg == :shutdown
            ws.close_websocket
          else
            ws.send msg
          end
        end
    
        ws.onmessage {|msg| @onmessage.call(msg); @channel2.push msg}

        ws.onerror {|e| @onerror.call(e)}

        ws.onclose {@onclose.call(ws); @channel1.unsubscribe sid}
      end
      EM.add_timer(0.1) {ready = true}
    end
  end
  sleep 0.2 until ready
  @websocket
end