Class: FluQ::Input::Socket

Inherits:
Base
  • Object
show all
Defined in:
lib/fluq/input/socket.rb

Constant Summary collapse

MAXLEN = 
16 * 1024

Instance Attribute Summary collapse

Attributes inherited from Base

#config, #worker

Instance Method Summary collapse

Methods inherited from Base

#process

Methods included from Mixins::Loggable

#logger

Constructor Details

#initialize ⇒ Socket

Constructor.

Examples:

Launch a server


server = FluQ::Server.new(runner, bind: "tcp://localhost:7654")

Parameters:

  • options (Hash) —

    a customizable set of options



15
16
17
 
# File 'lib/fluq/input/socket.rb', line 15

def initialize(*)
  super
end

Instance Attribute Details

#url ⇒ Object (readonly) 



5
6
7
 
# File 'lib/fluq/input/socket.rb', line 5

def url
  @url
end

Instance Method Details

#description ⇒ String

Returns descriptive name.

Returns:

  • (String) —

    descriptive name



25
26
27
 
# File 'lib/fluq/input/socket.rb', line 25

def description
  "socket (#{@url})"
end

#listening? ⇒ Boolean

Returns true when listening.

Returns:

  • (Boolean) —

    true when listening



51
52
53
 
# File 'lib/fluq/input/socket.rb', line 51

def listening?
  !!@server
end

#name ⇒ String

Returns short name.

Returns:

  • (String) —

    short name



20
21
22
 
# File 'lib/fluq/input/socket.rb', line 20

def name
  @url ? @url.scheme : super
end

#run ⇒ Object

Start the server



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/fluq/input/socket.rb', line 30

def run
  super

  @server = case @url.scheme
  when 'tcp'
    TCPServer.new(@url.host, @url.port)
  when 'udp'
    UDPSocket.new.tap {|s| s.bind(@url.host, @url.port) }
  when 'unix'
    UNIXServer.open(@url.path)
  end

  case @url.scheme
  when 'udp'
    loop { process @server.recvfrom(MAXLEN)[0] }
  else
    loop { async.handle_connection @server.accept }
  end
end