Module: MaZMQ::Proxy::Backend

Included in:
MaZMQ::Proxy
Defined in:
lib/ma-zmq/proxy/backend.rb

Instance Method Summary collapse

Instance Method Details

#connect(protocol, address, port) ⇒ Object 



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/ma-zmq/proxy/backend.rb', line 10

def connect(protocol, address, port)
  # validate as in SocketHandler
  request = MaZMQ::Request.new
  request.connect(protocol, address, port)
  if EM.reactor_running?
    request.timeout(@timeout)
    if @on_read_lambda.is_a? Proc
      request.on_read { |msg|
        @on_read_lambda.call(msg)
      }
    end
    if @on_timeout_lambda.is_a? Proc
      request.on_timeout {
        @on_timeout_lambda.call
      }
    end
  end

  #request.identity = "lb-#{@@id}"
  #@@id += 1
  @sockets << request
  @sockets.size - 1
end

#disconnect(index) ⇒ Object

def reconnect(index) end



37
38
39
40
 
# File 'lib/ma-zmq/proxy/backend.rb', line 37

def disconnect(index)
  socket = @sockets.delete_at(index)
  socket.close
end

#initialize ⇒ Object 



4
5
6
7
8
 
# File 'lib/ma-zmq/proxy/backend.rb', line 4

def initialize
  @sockets = []

  @timeout = nil # TODO individual timeouts for different sockets
end

#on_read(&block) ⇒ Object 



59
60
61
62
63
64
65
66
67
 
# File 'lib/ma-zmq/proxy/backend.rb', line 59

def on_read(&block)
  return false if not EM.reactor_running?
  @on_read_lambda = block
  @sockets.each do |socket|
    socket.on_read { |msg|
      @on_read_lambda.call(msg)
    }
  end
end

#on_timeout(&block) ⇒ Object 



49
50
51
52
53
54
55
56
57
 
# File 'lib/ma-zmq/proxy/backend.rb', line 49

def on_timeout(&block)  
  return false if not EM.reactor_running?
  @on_timeout_lambda = block
  @sockets.each do |socket|
    socket.on_timeout {
      @on_timeout_lambda.call
    }
  end
end

#timeout(secs) ⇒ Object 



42
43
44
45
46
47
 
# File 'lib/ma-zmq/proxy/backend.rb', line 42

def timeout(secs)
  @timeout = secs
  @sockets.each do |s|
    s.timeout @timeout
  end
end