lib/rbot/ircsocket.rb

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