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