Class: Fargo::Connection::Base

Inherits:
Object
  • Object
show all
Includes:
ActiveSupport::Callbacks, ActiveSupport::Configurable, Publisher
Defined in:
lib/fargo/connection/base.rb

Direct Known Subclasses

Download, Hub, Search, Upload

Instance Attribute Summary collapse

Attributes included from Publisher

#subscribers

Instance Method Summary collapse

Methods included from Publisher

#publish, #subscribe, #subscribed_to?, #unsubscribe

Constructor Details

#initialize(client) ⇒ Base

Returns a new instance of Base.



17
18
19
20
21
 
# File 'lib/fargo/connection/base.rb', line 17

def initialize client
  @outgoing = Queue.new
  @client   = client
  config.quit_on_disconnect = true
end

Instance Attribute Details

#socket ⇒ Object

Returns the value of attribute socket.



14
15
16
 
# File 'lib/fargo/connection/base.rb', line 14

def socket
  @socket
end

Instance Method Details

#connect ⇒ Object 



23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/fargo/connection/base.rb', line 23

def connect
  Fargo.logger.info(
    "#{self}: Opening connection with #{config.address}, #{config.port}"
  )

  open_socket
  listen

  connection_type = self.class.name.split('::').last.downcase
  @client.publish :"#{connection_type}_connection_opened"
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


45
46
47
 
# File 'lib/fargo/connection/base.rb', line 45

def connected?
  !@socket.nil?
end

#disconnect ⇒ Object 



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/fargo/connection/base.rb', line 65

def disconnect
  Fargo.logger.debug "#{self}: Disconnecting connection"

  write "$Quit #{@client.config.nick}" if config.quit_on_disconnect

  if @threads
    @threads.each &:exit
    @threads.clear
  end

  if @socket
    begin
      @socket.close
    rescue => e
      Fargo.logger.error "Error closing socket: #{e}"
    end
  end

  @socket = nil
  @outgoing.clear

  connection_type = self.class.name.split('::').last.downcase
  @client.publish :"#{connection_type}_disconnected"
end

#listen ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/fargo/connection/base.rb', line 49

def listen
  return unless @threads.nil? || @threads.size == 0
  
  run_callbacks :listen do
    @threads = []

    # Start a thread to read the socket
    @threads << Thread.start { loop { read_data } }

    # Start a thread to send information from the queue
    @threads << Thread.start { loop { write_data @outgoing.pop } }

    @threads.each { |t| t.abort_on_exception = true }
  end
end

#open_socket ⇒ Object 



39
40
41
42
43
 
# File 'lib/fargo/connection/base.rb', line 39

def open_socket
  @socket ||= TCPSocket.open config.address, config.port
rescue Errno::ECONNREFUSED
  raise Fargo::ConnectionError.new "Couldn't open a connection to #{config.address}:#{config.port}"
end

#receive ⇒ Object 



35
36
37
 
# File 'lib/fargo/connection/base.rb', line 35

def receive
  raise 'Implement me!'
end

#write(string) ⇒ Object 



90
91
92
93
94
 
# File 'lib/fargo/connection/base.rb', line 90

def write string
  string << '|' unless string.end_with?('|')
  @outgoing << string # append this to the queue of things to be written
  true
end