Utils: convert unknown numering HTML entities to the corresponding UTF-8 characters...
[rbot] / lib / rbot / ircsocket.rb
1 class ::String
2   # Calculate the penalty which will be assigned to this message
3   # by the IRCd
4   def irc_send_penalty
5     # According to eggrdop, the initial penalty is
6     penalty = 1 + self.size/100
7     # on everything but UnderNET where it's
8     # penalty = 2 + self.size/120
9
10     cmd, pars = self.split($;,2)
11     debug "cmd: #{cmd}, pars: #{pars.inspect}"
12     case cmd.to_sym
13     when :KICK
14       chan, nick, msg = pars.split
15       chan = chan.split(',')
16       nick = nick.split(',')
17       penalty += nick.size
18       penalty *= chan.size
19     when :MODE
20       chan, modes, argument = pars.split
21       extra = 0
22       if modes
23         extra = 1
24         if argument
25           extra += modes.split(/\+|-/).size
26         else
27           extra += 3 * modes.split(/\+|-/).size
28         end
29       end
30       if argument
31         extra += 2 * argument.split.size
32       end
33       penalty += extra * chan.split.size
34     when :TOPIC
35       penalty += 1
36       penalty += 2 unless pars.split.size < 2
37     when :PRIVMSG, :NOTICE
38       dests = pars.split($;,2).first
39       penalty += dests.split(',').size
40     when :WHO
41       # I'm too lazy to implement this one correctly
42       penalty += 5
43     when :AWAY, :JOIN, :VERSION, :TIME, :TRACE, :WHOIS, :DNS
44       penalty += 2
45     when :INVITE, :NICK
46       penalty += 3
47     when :ISON
48       penalty += 1
49     else # Unknown messages
50       penalty += 1
51     end
52     if penalty > 99
53       debug "Wow, more than 99 secs of penalty!"
54       penalty = 99
55     end
56     if penalty < 2
57       debug "Wow, less than 2 secs of penalty!"
58       penalty = 2
59     end
60     debug "penalty: #{penalty}"
61     return penalty
62   end
63 end
64
65 module Irc
66
67   require 'socket'
68   require 'thread'
69   require 'rbot/timer'
70
71   class QueueRing
72     # A QueueRing is implemented as an array with elements in the form
73     # [chan, [message1, message2, ...]
74     # Note that the channel +chan+ has no actual bearing with the channels
75     # to which messages will be sent
76
77     def initialize
78       @storage = Array.new
79       @last_idx = -1
80     end
81
82     def clear
83       @storage.clear
84       @last_idx = -1
85     end
86
87     def length
88       len = 0
89       @storage.each {|c|
90         len += c[1].size
91       }
92       return len
93     end
94     alias :size :length
95
96     def empty?
97       @storage.empty?
98     end
99
100     def push(mess, chan)
101       cmess = @storage.assoc(chan)
102       if cmess
103         idx = @storage.index(cmess)
104         cmess[1] << mess
105         @storage[idx] = cmess
106       else
107         @storage << [chan, [mess]]
108       end
109     end
110
111     def next
112       if empty?
113         warning "trying to access empty ring"
114         return nil
115       end
116       save_idx = @last_idx
117       @last_idx = (@last_idx + 1) % @storage.size
118       mess = @storage[@last_idx][1].first
119       @last_idx = save_idx
120       return mess
121     end
122
123     def shift
124       if empty?
125         warning "trying to access empty ring"
126         return nil
127       end
128       @last_idx = (@last_idx + 1) % @storage.size
129       mess = @storage[@last_idx][1].shift
130       @storage.delete(@storage[@last_idx]) if @storage[@last_idx][1] == []
131       return mess
132     end
133
134   end
135
136   class MessageQueue
137     def initialize
138       # a MessageQueue is an array of QueueRings
139       # rings have decreasing priority, so messages in ring 0
140       # are more important than messages in ring 1, and so on
141       @rings = Array.new(3) { |i|
142         if i > 0
143           QueueRing.new
144         else
145           # ring 0 is special in that if it's not empty, it will
146           # be popped. IOW, ring 0 can starve the other rings
147           # ring 0 is strictly FIFO and is therefore implemented
148           # as an array
149           Array.new
150         end
151       }
152       # the other rings are satisfied round-robin
153       @last_ring = 0
154     end
155
156     def clear
157       @rings.each { |r|
158         r.clear
159       }
160       @last_ring = 0
161     end
162
163     def push(mess, chan=nil, cring=0)
164       ring = cring
165       if ring == 0
166         warning "message #{mess} at ring 0 has channel #{chan}: channel will be ignored" if !chan.nil?
167         @rings[0] << mess
168       else
169         error "message #{mess} at ring #{ring} must have a channel" if chan.nil?
170         @rings[ring].push mess, chan
171       end
172     end
173
174     def empty?
175       @rings.each { |r|
176         return false unless r.empty?
177       }
178       return true
179     end
180
181     def length
182       len = 0
183       @rings.each { |r|
184         len += r.size
185       }
186       len
187     end
188     alias :size :length
189
190     def next
191       if empty?
192         warning "trying to access empty ring"
193         return nil
194       end
195       mess = nil
196       if !@rings[0].empty?
197         mess = @rings[0].first
198       else
199         save_ring = @last_ring
200         (@rings.size - 1).times {
201           @last_ring = (@last_ring % (@rings.size - 1)) + 1
202           if !@rings[@last_ring].empty?
203             mess = @rings[@last_ring].next
204             break
205           end
206         }
207         @last_ring = save_ring
208       end
209       error "nil message" if mess.nil?
210       return mess
211     end
212
213     def shift
214       if empty?
215         warning "trying to access empty ring"
216         return nil
217       end
218       mess = nil
219       if !@rings[0].empty?
220         return @rings[0].shift
221       end
222       (@rings.size - 1).times {
223         @last_ring = (@last_ring % (@rings.size - 1)) + 1
224         if !@rings[@last_ring].empty?
225           return @rings[@last_ring].shift
226         end
227       }
228       error "nil message" if mess.nil?
229       return mess
230     end
231
232   end
233
234   # wrapped TCPSocket for communication with the server.
235   # emulates a subset of TCPSocket functionality
236   class IrcSocket
237
238     MAX_IRC_SEND_PENALTY = 10
239
240     # total number of lines sent to the irc server
241     attr_reader :lines_sent
242
243     # total number of lines received from the irc server
244     attr_reader :lines_received
245
246     # total number of bytes sent to the irc server
247     attr_reader :bytes_sent
248
249     # total number of bytes received from the irc server
250     attr_reader :bytes_received
251
252     # accumulator for the throttle
253     attr_reader :throttle_bytes
254
255     # delay between lines sent
256     attr_reader :sendq_delay
257
258     # max lines to burst
259     attr_reader :sendq_burst
260
261     # an optional filter object. we call @filter.in(data) for
262     # all incoming data and @filter.out(data) for all outgoing data
263     attr_reader :filter
264
265     # normalized uri of the current server
266     attr_reader :server_uri
267
268     # default trivial filter class
269     class IdentityFilter
270         def in(x)
271             x
272         end
273
274         def out(x)
275             x
276         end
277     end
278
279     # set filter to identity, not to nil
280     def filter=(f)
281         @filter = f || IdentityFilter.new
282     end
283
284     # server_list:: list of servers to connect to
285     # host::   optional local host to bind to (ruby 1.7+ required)
286     # create a new IrcSocket
287     def initialize(server_list, host, sendq_delay=2, sendq_burst=4, opts={})
288       @timer = Timer::Timer.new
289       @timer.add(0.2) do
290         spool
291       end
292       @server_list = server_list.dup
293       @server_uri = nil
294       @conn_count = 0
295       @host = host
296       @sock = nil
297       @filter = IdentityFilter.new
298       @spooler = false
299       @lines_sent = 0
300       @lines_received = 0
301       if opts.kind_of?(Hash) and opts.key?(:ssl)
302         @ssl = opts[:ssl]
303       else
304         @ssl = false
305       end
306
307       if sendq_delay
308         @sendq_delay = sendq_delay.to_f
309       else
310         @sendq_delay = 2
311       end
312       @last_send = Time.new - @sendq_delay
313       @flood_send = Time.new
314       @last_throttle = Time.new
315       @burst = 0
316       if sendq_burst
317         @sendq_burst = sendq_burst.to_i
318       else
319         @sendq_burst = 4
320       end
321     end
322
323     def connected?
324       !@sock.nil?
325     end
326
327     # open a TCP connection to the server
328     def connect
329       if connected?
330         warning "reconnecting while connected"
331         return
332       end
333       srv_uri = @server_list[@conn_count % @server_list.size].dup
334       srv_uri = 'irc://' + srv_uri if !(srv_uri =~ /:\/\//)
335       @conn_count += 1
336       @server_uri = URI.parse(srv_uri)
337       @server_uri.port = 6667 if !@server_uri.port
338       debug "connection attempt \##{@conn_count} (#{@server_uri.host}:#{@server_uri.port})"
339
340       if(@host)
341         begin
342           @sock=TCPSocket.new(@server_uri.host, @server_uri.port, @host)
343         rescue ArgumentError => e
344           error "Your version of ruby does not support binding to a "
345           error "specific local address, please upgrade if you wish "
346           error "to use HOST = foo"
347           error "(this option has been disabled in order to continue)"
348           @sock=TCPSocket.new(@server_uri.host, @server_uri.port)
349         end
350       else
351         @sock=TCPSocket.new(@server_uri.host, @server_uri.port)
352       end
353       if(@ssl)
354         require 'openssl'
355         ssl_context = OpenSSL::SSL::SSLContext.new()
356         ssl_context.verify_mode = OpenSSL::SSL::VERIFY_NONE
357         @rawsock = @sock
358         @sock = OpenSSL::SSL::SSLSocket.new(@rawsock, ssl_context)
359         @sock.sync_close = true
360         @sock.connect
361       end
362       @qthread = false
363       @qmutex = Mutex.new
364       @sendq = MessageQueue.new
365     end
366
367     def sendq_delay=(newfreq)
368       debug "changing sendq frequency to #{newfreq}"
369       @qmutex.synchronize do
370         @sendq_delay = newfreq
371         if newfreq == 0
372           clearq
373           @timer.stop
374         else
375           @timer.start
376         end
377       end
378     end
379
380     def sendq_burst=(newburst)
381       @qmutex.synchronize do
382         @sendq_burst = newburst
383       end
384     end
385
386     # used to send lines to the remote IRCd by skipping the queue
387     # message: IRC message to send
388     # it should only be used for stuff that *must not* be queued,
389     # i.e. the initial PASS, NICK and USER command
390     # or the final QUIT message
391     def emergency_puts(message)
392       @qmutex.synchronize do
393         # debug "In puts - got mutex"
394         puts_critical(message)
395       end
396     end
397
398     def handle_socket_error(string, err)
399       error "#{string} failed: #{err.inspect}"
400       debug err.backtrace.join("\n")
401       # We assume that an error means that there are connection
402       # problems and that we should reconnect, so we
403       shutdown
404       raise SocketError.new(err.inspect)
405     end
406
407     # get the next line from the server (blocks)
408     def gets
409       if @sock.nil?
410         warning "socket get attempted while closed"
411         return nil
412       end
413       begin
414         reply = @filter.in(@sock.gets)
415         @lines_received += 1
416         reply.strip! if reply
417         debug "RECV: #{reply.inspect}"
418         return reply
419       rescue => e
420         handle_socket_error(:RECV, e)
421       end
422     end
423
424     def queue(msg, chan=nil, ring=0)
425       if @sendq_delay > 0
426         @qmutex.synchronize do
427           @sendq.push msg, chan, ring
428           @timer.start
429         end
430       else
431         # just send it if queueing is disabled
432         self.emergency_puts(msg)
433       end
434     end
435
436     # pop a message off the queue, send it
437     def spool
438       @qmutex.synchronize do
439         begin
440           debug "in spooler"
441           if @sendq.empty?
442             @timer.stop
443             return
444           end
445           now = Time.new
446           if (now >= (@last_send + @sendq_delay))
447             debug "resetting @burst"
448             @burst = 0
449           elsif (@burst > @sendq_burst)
450             # nope. can't send anything, come back to us next tick...
451             debug "can't send yet"
452             @timer.start
453             return
454           end
455           @flood_send = now if @flood_send < now
456           debug "can send #{@sendq_burst - @burst} lines, there are #{@sendq.size} to send"
457           while !@sendq.empty? and @burst < @sendq_burst and @flood_send - now < MAX_IRC_SEND_PENALTY
458             debug "sending message (#{@flood_send - now} < #{MAX_IRC_SEND_PENALTY})"
459             puts_critical(@sendq.shift, true)
460           end
461           if @sendq.empty?
462             @timer.stop
463           end
464         rescue => e
465           error "Spooling failed: #{e.inspect}"
466           error e.backtrace.join("\n")
467         end
468       end
469     end
470
471     def clearq
472       if @sock
473         @qmutex.synchronize do
474           unless @sendq.empty?
475             @sendq.clear
476           end
477         end
478       else
479         warning "Clearing socket while disconnected"
480       end
481     end
482
483     # flush the TCPSocket
484     def flush
485       @sock.flush
486     end
487
488     # Wraps Kernel.select on the socket
489     def select(timeout=nil)
490       Kernel.select([@sock], nil, nil, timeout)
491     end
492
493     # shutdown the connection to the server
494     def shutdown(how=2)
495       return unless connected?
496       begin
497         @sock.close
498       rescue => err
499         error "error while shutting down: #{err.inspect}"
500         debug err.backtrace.join("\n")
501       end
502       @rawsock = nil if @ssl
503       @sock = nil
504       @burst = 0
505     end
506
507     private
508
509     # same as puts, but expects to be called with a mutex held on @qmutex
510     def puts_critical(message, penalty=false)
511       # debug "in puts_critical"
512       begin
513         debug "SEND: #{message.inspect}"
514         if @sock.nil?
515           error "SEND attempted on closed socket"
516         else
517           @sock.puts(@filter.out(message))
518           @last_send = Time.new
519           @flood_send += message.irc_send_penalty if penalty
520           @lines_sent += 1
521           @burst += 1
522         end
523       rescue => e
524         handle_socket_error(:SEND, e)
525       end
526     end
527
528   end
529
530 end