lib/rbot/ircsocket.rb

   1 module Irc
   2
   3   require 'socket'
   4   require 'thread'
   5   require 'rbot/timer'
   6
   7   # wrapped TCPSocket for communication with the server.
   8   # emulates a subset of TCPSocket functionality
   9   class IrcSocket
  10     # total number of lines sent to the irc server
  11     attr_reader :lines_sent
  12
  13     # total number of lines received from the irc server
  14     attr_reader :lines_received
  15
  16     # delay between lines sent
  17     attr_reader :sendq_delay
  18
  19     # max lines to burst
  20     attr_reader :sendq_burst
  21
  22     # server:: server to connect to
  23     # port::   IRCd port
  24     # host::   optional local host to bind to (ruby 1.7+ required)
  25     # create a new IrcSocket
  26     def initialize(server, port, host, sendq_delay=2, sendq_burst=4)
  27       @timer = Timer::Timer.new
  28       @timer.add(0.2) do
  29         spool
  30       end
  31       @server = server.dup
  32       @port = port.to_i
  33       @host = host
  34       @spooler = false
  35       @lines_sent = 0
  36       @lines_received = 0
  37       if sendq_delay
  38         @sendq_delay = sendq_delay.to_f
  39       else
  40         @sendq_delay = 2
  41       end
  42       @last_send = Time.new - @sendq_delay
  43       @burst = 0
  44       if sendq_burst
  45         @sendq_burst = sendq_burst.to_i
  46       else
  47         @sendq_burst = 4
  48       end
  49     end
  50
  51     # open a TCP connection to the server
  52     def connect
  53       if(@host)
  54         begin
  55           @sock=TCPSocket.new(@server, @port, @host)
  56         rescue ArgumentError => e
  57           $stderr.puts "Your version of ruby does not support binding to a "
  58           $stderr.puts "specific local address, please upgrade if you wish "
  59           $stderr.puts "to use HOST = foo"
  60           $stderr.puts "(this option has been disabled in order to continue)"
  61           @sock=TCPSocket.new(@server, @port)
  62         end
  63       else
  64         @sock=TCPSocket.new(@server, @port)
  65       end
  66       @qthread = false
  67       @qmutex = Mutex.new
  68       @sendq = Array.new
  69     end
  70
  71     def sendq_delay=(newfreq)
  72       debug "changing sendq frequency to #{newfreq}"
  73       @qmutex.synchronize do
  74         @sendq_delay = newfreq
  75         if newfreq == 0
  76           clearq
  77           @timer.stop
  78         else
  79           @timer.start
  80         end
  81       end
  82     end
  83
  84     def sendq_burst=(newburst)
  85       @qmutex.synchronize do
  86         @sendq_burst = newburst
  87       end
  88     end
  89
  90     # used to send lines to the remote IRCd
  91     # message: IRC message to send
  92     def puts(message)
  93       @qmutex.synchronize do
  94         # debug "In puts - got mutex"
  95         puts_critical(message)
  96       end
  97     end
  98
  99     # get the next line from the server (blocks)
 100     def gets
 101       reply = @sock.gets
 102       @lines_received += 1
 103       reply.strip! if reply
 104       debug "RECV: #{reply.inspect}"
 105       reply
 106     end
 107
 108     def queue(msg)
 109       if @sendq_delay > 0
 110         @qmutex.synchronize do
 111           @sendq.push msg
 112         end
 113         @timer.start
 114       else
 115         # just send it if queueing is disabled
 116         self.puts(msg)
 117       end
 118     end
 119
 120     # pop a message off the queue, send it
 121     def spool
 122       if @sendq.empty?
 123         @timer.stop
 124         return
 125       end
 126       now = Time.new
 127       if (now >= (@last_send + @sendq_delay))
 128         # reset burst counter after @sendq_delay has passed
 129         @burst = 0
 130         debug "in spool, resetting @burst"
 131       elsif (@burst >= @sendq_burst)
 132         # nope. can't send anything, come back to us next tick...
 133         @timer.start
 134         return
 135       end
 136       @qmutex.synchronize do
 137         debug "(can send #{@sendq_burst - @burst} lines, there are #{@sendq.length} to send)"
 138         (@sendq_burst - @burst).times do
 139           break if @sendq.empty?
 140           puts_critical(@sendq.shift)
 141         end
 142       end
 143       if @sendq.empty?
 144         @timer.stop
 145       end
 146     end
 147
 148     def clearq
 149       unless @sendq.empty?
 150         @qmutex.synchronize do
 151           @sendq.clear
 152         end
 153       end
 154     end
 155
 156     # flush the TCPSocket
 157     def flush
 158       @sock.flush
 159     end
 160
 161     # Wraps Kernel.select on the socket
 162     def select(timeout=nil)
 163       Kernel.select([@sock], nil, nil, timeout)
 164     end
 165
 166     # shutdown the connection to the server
 167     def shutdown(how=2)
 168       @sock.shutdown(how)
 169     end
 170
 171     private
 172
 173     # same as puts, but expects to be called with a mutex held on @qmutex
 174     def puts_critical(message)
 175       # debug "in puts_critical"
 176       debug "SEND: #{message.inspect}"
 177       @sock.send(message + "\n",0)
 178       @last_send = Time.new
 179       @lines_sent += 1
 180       @burst += 1
 181     end
 182
 183   end
 184
 185 end