Class: RServiceBus2::Transporter

Inherits:
Object
  • Object
show all
Defined in:
lib/rservicebus2/transporter.rb

Overview

TODO: Poison Message? Can I bury with timeout in beanstalk ? Needs to end up on an error queue, destination queue may be down.

Instance Method Summary collapse

Instance Method Details

#connect(remote_host_name) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/rservicebus2/transporter.rb', line 53

def connect(remote_host_name)
  RServiceBus2.rlog "connect called, #{remote_host_name}"
  if @gateway.nil? || remoteHostName != @remote_host_name || @destination.nil?
    disconnect
  end

  if @gateway.nil?
    # Get destination url from job
    @remote_host_name = remote_host_name
    @remote_user_name = get_value("REMOTE_USER_#{remote_host_name.upcase}")
    if @remote_user_name.nil?
      RServiceBus2.log "**** Username not specified for Host, #{remoteHostName}"
      RServiceBus2.log "**** Add an environment variable of the form, REMOTE_USER_#{remoteHostName.upcase}=[USERNAME]"
      abort
    end

    @local_port = get_value('LOCAL_PORT', 27018).to_i
    RServiceBus2.rlog "Local Port: #{@local_port}"
    begin
      RServiceBus2.log "Connect SSH, #{@remote_user_name}@#{@remoteHostName}"
      # Open port 27018 to forward to 127.0.0.11300 on the remote host
      @gateway = Net::SSH::Gateway.new(@remote_host_name, @remote_user_name)
      @gateway.open('127.0.0.1', 11300, @local_port)
      RServiceBus2.log "Connected to SSH, #{@remote_user_name}@#{@remote_host_name}"

    rescue Errno::EADDRINUSE
      puts "*** Local transport port in use, #{@local_port}"
      puts "*** Change local transport port, #{@localPort}, using format, LOCAL_PORT=#{@localPort+1}"
      abort
    rescue Errno::EACCES
      puts "*** Local transport port specified, #{@local_port}, needs sudo access"
      puts '*** Change local transport port using format, LOCAL_PORT=27018'
      abort
    end

    begin
      destination_url = "127.0.0.1:#{@local_port}"
      RServiceBus2.rlog "Connect to Remote Beanstalk, #{destination_url}"
      @destination = Beanstalk::Pool.new([destinationUrl])
      RServiceBus2.rlog "Connected to Remote Beanstalk, #{destination_url}"
    rescue StandardError => e
      if e.message == 'Beanstalk::NotConnected'
        puts "***Could not connect to destination, check beanstalk is running at, #{destination_url}"
        raise CouldNotConnectToDestination
      end
      raise
    end
  end
end

#connect_to_source_beanstalk ⇒ Object 



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/rservicebus2/transporter.rb', line 18

def connect_to_source_beanstalk
  source_queue_name = get_value('SOURCE_QUEUE_NAME', 'transport-out')
  source_url = get_value('SOURCE_URL', '127.0.0.1:11300')
  @source = Beanstalk::Pool.new([source_url])
  @source.watch source_queue_name

  RServiceBus2.log "Connected to, #{source_queue_name}@#{source_url}"

rescue StandardError => e
  puts 'Error connecting to Beanstalk'
  puts "Host string, #{sourceUrl}"
  if e.message == 'Beanstalk::NotConnected'
    puts '***Most likely, beanstalk is not running. Start beanstalk, and try running this again.'
    puts "***If you still get this error, check beanstalk is running at, #{sourceUrl}"
  else
    puts e.message
    puts e.backtrace
  end
  abort
end

#disconnect ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/rservicebus2/transporter.rb', line 39

def disconnect
  RServiceBus2.log "Disconnect from,
    #{@remote_user_name}@#{@remote_host_name}/#{@remote_queue_name}"
  @gateway.shutdown! unless @gateway.nil?
  @gateway = nil
  @remote_host_name = nil

  @destination.close unless @destination.nil?
  @destination = nil

  @remote_user_name = nil
  @remote_queue_name = nil
end

#get_value(name, default = nil) ⇒ Object 



12
13
14
15
16
 
# File 'lib/rservicebus2/transporter.rb', line 12

def get_value(name, default = nil)
  value = (ENV[name].nil? || ENV[name] == '') ? default : ENV[name]
  RServiceBus2.log "Env value: #{name}: #{value}"
  value
end

#process ⇒ Object 



103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/rservicebus2/transporter.rb', line 103

def process
  # Get the next job from the source queue
  job = @source.reserve @timeout
  msg = YAML.load(job.body)

  connect(msg.remote_host_name)

  @remote_queue_name = msg.remote_queue_name
  RServiceBus2.rlog "Put msg, #{msg.remote_queue_name}"
  @destination.use(msg.remote_queue_name)
  @destination.put(job.body)
  RServiceBus2.log "Msg put, #{msg.remote_queue_name}"

  unless ENV['AUDIT_QUEUE_NAME'].nil?
    @source.use ENV['AUDIT_QUEUE_NAME']
    @source.put job.body
  end
  # remove job
  job.delete

  RServiceBus2.log "Job sent to, #{@remote_user_name}@#{@remote_host_name}/#{@remote_queue_name}"
rescue StandardError => e
  disconnect
  if e.message == 'TIMED_OUT'
    RServiceBus2.rlog 'No Msg'
    return
  end
  raise e
end

#run ⇒ Object 



133
134
135
136
137
138
139
140
 
# File 'lib/rservicebus2/transporter.rb', line 133

def run
  @timeout = get_value('TIMEOUT', 5)
  connectToSourceBeanstalk
  while true
    process
  end
  disconnect_from_remote_ssh
end