Module: Smash::CloudPowers::Synapse::Pipe

Included in:
Smash::CloudPowers::SelfAwareness, Smash::CloudPowers::Synapse
Defined in:
lib/cloud_powers/synapse/pipe.rb

Instance Method Summary collapse

Instance Method Details

#create_stream(name) ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 9

def create_stream(name)
  begin
    config = stream_config(stream_name: env(name))
    resp = kinesis.create_stream(config)
    kinesis.wait_until(:stream_exists, stream_name: config[:stream_name])
    resp.successful? # (http request successful && stream created)?
  rescue Exception => e
    if e.kind_of? Aws::Kinesis::Errors::ResourceInUseException
      logger.info "#{name} already created"
      stream_status = kinesis.describe_stream(name).stream_description.stream_status
      return if stream_status == 'ACTIVE'
      logger.info "Not ready for traffic.  Wait for 30 seconds..."
      sleep 30
      nil # no request -> no response
    else
      # TODO: make the errors thing work
      error_message = format_error_message(e)
      logger.error error_message
      false # the request was not successful
    end
  end
end

#flow_from_pipe(stream) ⇒ Object 



32
33
34
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 32

def flow_from_pipe(stream)
  throw NotImplementedError
end

#flow_to_pipe(stream) ⇒ Object 



36
37
38
39
40
41
42
43
44
45
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 36

def flow_to_pipe(stream)
  create_stream(stream) unless stream_exists? stream
  records = yield if block_given?
  body = message_body_collection(records)
  # TODO: this isn't working yet.  figure out retry logic
  # resp = kinesis.put_records(body)
  # retry(lambda { stream_exists? stream }) flow_to(stream)
  @last_sequence_number = resp.records.map(&:sequence_number).sort.last
  # TODO: what to return? true?
end

#from_pipe(stream) ⇒ Object 



47
48
49
50
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 47

def from_pipe(stream)
  # implemented get_records and/or other consuming app stuff
  throw NotImplementedError
end

#message_body_collection(records) ⇒ Object 



52
53
54
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 52

def message_body_collection(records)
  throw NotImplementedError
end

#pipe_message_body(opts = {}) ⇒ Object 



56
57
58
59
60
61
62
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 56

def pipe_message_body(opts = {})
  {
    stream_name:      env(opts[:stream_name]) || env('status_stream'),
    data:             opts[:data] || update_message_body(opts),
    partition_key:    opts[:partition_key] || @instance_id
  }
end

#pipe_to(stream) ⇒ Object 



64
65
66
67
68
69
70
71
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 64

def pipe_to(stream)
  create_stream(stream) unless stream_exists? stream
  message = yield if block_given?
  body = update_message_body(message)
  resp = kinesis.put_record pipe_message_body(stream_name: stream, data: body.to_json)
  # TODO: implement retry logic for failed request
  @last_sequence_number = resp.sequence_number
end

#stream_config(opts = {}) ⇒ Object 



73
74
75
76
77
78
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 73

def stream_config(opts = {})
  config = {
    stream_name: opts[:stream_name] || env('status_stream'),
    shard_count: opts[:shard_count] || 1
  }
end

#stream_exists?(name) ⇒ Boolean

Returns:

  • (Boolean) 


80
81
82
83
84
85
86
 
# File 'lib/cloud_powers/synapse/pipe.rb', line 80

def stream_exists?(name)
  begin
    kinesis.describe_stream(stream_name: env(name))
  rescue Aws::Kinesis::Errors::ResourceNotFoundException => e
    false
  end
end