aboutsummaryrefslogtreecommitdiff
path: root/lib/mauve/udp_server.rb
blob: 09afa01917d8edb7cbee68cd9e6367466ccce0dc (plain)
1
2
3
4
5
6
7
8
9
10
11
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
37
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
# encoding: UTF-8
require 'yaml'
require 'socket'
require 'mauve/datamapper'
require 'mauve/proto'
require 'mauve/alert'
require 'mauve/mauve_thread'
require 'ipaddr'

module Mauve

  #
  # This is the thread that accepts the packets.
  #
  class UDPServer < MauveThread

    include Singleton

    attr_reader   :ip, :port

    # Yup.  Creates a new UDPServer.
    #
    # Defaults:
    #   * listening IP: 127.0.0.1
    #   * listening port: 32741
    #
    def initialize
      #
      # Set up some defaults.
      #
      self.ip     = "127.0.0.1"
      self.port   = 32741
      @socket = nil

      super
    end

    #
    # This sets the IP which the server will listen on.
    #
    # @param [String] i The new IP
    # @return [IPAddr]
    #
    def ip=(i)
      raise ArgumentError, "ip must be a string" unless i.is_a?(String)
      @ip = IPAddr.new(i)
    end

    # Sets the listening port
    #
    # @param [Integer] pr The new port
    # @return [Integer]
    #
    def port=(pr)
      raise ArgumentError, "port must be an integer between 0 and #{2**16-1}" unless pr.is_a?(Integer) and pr < 2**16 and pr > 0
      @port = pr
    end

    # This stops the UDP server, by signalling to the thread to stop, and
    # sending a zero-length packet to the socket.
    #
    def stop
      @stop = true
      #
      # Triggers loop to close socket.
      #
      UDPSocket.open(Socket.const_get(@socket.addr[0])).send("", 0, @socket.addr[2], @socket.addr[1]) unless @socket.nil? or @socket.closed?

      super
    end

    private

    #
    # This opens the socket for listening.
    #
    # It tries to increase the default receiving buffer to 10M, and will warn
    # if this fails to increase the buffer at all.
    #
    # @return [Nilclass]
    #
    def open_socket
      #
      # Specify the family when opening the socket.
      #
      @socket = UDPSocket.new(@ip.family)

      logger.debug("Trying to increase Socket::SO_RCVBUF to 10M.")
      old = @socket.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).unpack("i").first

      @socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 10*1024*1024)
      new = @socket.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).unpack("i").first

      logger.warn "Could not increase Socket::SO_RCVBUF.  Had #{old} ended up with #{new}!" if old > new

      logger.debug("Successfully increased Socket::SO_RCVBUF from #{old} to #{new}.")

      @socket.bind(@ip.to_s, @port)

      logger.info("Opened socket on #{@ip.to_s}:#{@port}")
    end

    # This closes the socket.  IOErrors are caught and logged.
    #
    def close_socket
      return if @socket.nil?  or @socket.closed?

      begin
        @socket.close
      rescue IOError => ex
        # Just in case there is some sort of explosion!
        logger.error "Caught IOError #{ex.to_s}"
        logger.debug ex.backtrace.join("\n")
      end

      logger.info("Closed socket")
    end

    # This is the main loop.  It receives from the UDP port, and records the
    # time the packet arrives.  It then pushes the packet onto the Server's
    # packet buffer for the processor to pick up later.
    #
    # If a zero-length packet is received, and the thread has been signalled to
    # stop, then the socket is closed, and the thread exits.
    #
    def main_loop
      return if self.should_stop?

      open_socket if @socket.nil? or @socket.closed?

      return if self.should_stop?

      #
      # TODO: why is/isn't this non-block?
      #
      i = 0
      begin
#        packet      = @socket.recvfrom_nonblock(65535)
        packet      = @socket.recvfrom(65535)
        received_at = Time.now
      rescue Errno::EAGAIN, Errno::EWOULDBLOCK => ex
        IO.select([@socket])
        retry unless self.should_stop?
      end

      return if packet.nil?

      logger.debug("Got new packet: #{packet.inspect}")

      #
      # If we get a zero length packet, and we've been flagged to stop, we stop!
      #
      if packet.first.length == 0 and self.should_stop?
        close_socket
        return
      end

      #
      # Push packet onto central queue
      #
      Server.packet_push([packet[0], packet[1], received_at])
    end

  end

end