Class: ActiveElasticJob::Rack::SqsMessageConsumer

Inherits:
Object
  • Object
show all
Defined in:
lib/active_elastic_job/rack/sqs_message_consumer.rb

Overview

This middleware intercepts requests which are sent by the SQS daemon running in Amazon Elastic Beanstalk worker environments. It does this by looking at the User-Agent header. Requesets from the SQS daemon are handled in two alternative cases:

(1) the processed SQS message was originally triggered by a periodic task supported by Elastic Beanstalk’s Periodic Task feature

(2) the processed SQS message was queued by this gem representing an active job. In this case it verifies the digest which is sent along with a legit SQS message, and passed as an HTTP header in the resulting request. The digest is based on Rails’ secrets.secret_key_base. Therefore, the application running in the web environment, which generates the digest, and the application running in the worker environment, which verifies the digest, have to use the same secrets.secret_key_base setting.

Constant Summary collapse

OK_RESPONSE = 
[ '200'.freeze, { 'Content-Type'.freeze => 'text/plain'.freeze }, [ 'OK'.freeze ] ]
FORBIDDEN_RESPONSE = 
[
  '403'.freeze,
  { 'Content-Type'.freeze => 'text/plain'.freeze },
  [ 'Request forbidden!'.freeze ]
]
DOCKER_HOST_IP =

172.17.0.x is the default for Docker 172.18.0.x is the default for the bridge network of Docker Compose

/172.1(7|8).0.\d+/.freeze

Instance Method Summary collapse

Constructor Details

#initialize(app) ⇒ SqsMessageConsumer

:nodoc:



33
34
35
 
# File 'lib/active_elastic_job/rack/sqs_message_consumer.rb', line 33

def initialize(app) #:nodoc:
  @app = app
end

Instance Method Details

#call(env) ⇒ Object

:nodoc:



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/active_elastic_job/rack/sqs_message_consumer.rb', line 37

def call(env) #:nodoc:
  request = ActionDispatch::Request.new env
  if enabled? && (aws_sqsd?(request) || sqsd?(request))
    unless request.local? || sent_from_docker_host?(request)
      return FORBIDDEN_RESPONSE
    end

    if periodic_task?(request)
      execute_periodic_task(request)
      return OK_RESPONSE
    elsif originates_from_gem?(request)
      begin
        execute_job(request)
      rescue ActiveElasticJob::MessageVerifier::InvalidDigest => e
        return FORBIDDEN_RESPONSE
      end
      return OK_RESPONSE
    end
  end
  @app.call(env)
end