Class: Smash::CloudPowers::Synapse::Pipe::Stream
- Defined in:
- lib/cloud_powers/synapse/pipe/stream.rb
Instance Attribute Summary collapse
-
#kinesis ⇒ Object
Returns the value of attribute kinesis.
-
#shard_count ⇒ Object
Returns the value of attribute shard_count.
Attributes inherited from Resource
#call_name, #client, #linked, #meta, #name, #remote_id, #tags, #type
Instance Method Summary collapse
- #config ⇒ Object
- #create_resource ⇒ Object
-
#initialize(name:, client: kinesis, **config) ⇒ Stream
constructor
A new instance of Stream.
Methods included from Zenv
#env_vars, #i_vars, #lsof_cwd, #pid, #proc_cwd, #process_search, #project_root, #project_root=, #ps_cwd, #system_vars, #zfind, #zselect
Methods included from Helpers
#create_logger, #log_file, #logger
Methods included from PathHelp
#common_delimiter, #expand_path, #file_exists?, #file_search, #filename?, #job_exist?, #job_path, #job_require_path, #path_search, #paths_gcd, #paths_lcd, #to_path, #to_pathname, #to_realpath, #touch, #zlib_path
Methods included from LogicHelp
#attr_map, #called_from, #i_var_hash, #instance_attr_accessor, #smart_retry, #update_message_body, #wait_until
Methods included from LangHelp
#deep_modify_keys_with, #extract!, #find_and_remove, #format_error_message, #from_json, #modify_keys_with, #to_basic_hash, #to_camel, #to_hyph, #to_i_var, #to_pascal, #to_ruby_file_name, #to_snake, #valid_json?, #valid_url?
Methods included from AwsResources
#ec2, #image, #queue_poller, #region, #s3, #sns, #sqs
Methods included from Auth
Methods included from Creatable
Constructor Details
#initialize(name:, client: kinesis, **config) ⇒ Stream
Returns a new instance of Stream.
9 10 11 12 13 |
# File 'lib/cloud_powers/synapse/pipe/stream.rb', line 9 def initialize(name:, client: kinesis, **config) super @kinesis = client @shard_count = config[:shard_count] || 1 end |
Instance Attribute Details
#kinesis ⇒ Object
Returns the value of attribute kinesis.
7 8 9 |
# File 'lib/cloud_powers/synapse/pipe/stream.rb', line 7 def kinesis @kinesis end |
#shard_count ⇒ Object
Returns the value of attribute shard_count.
7 8 9 |
# File 'lib/cloud_powers/synapse/pipe/stream.rb', line 7 def shard_count @shard_count end |
Instance Method Details
#config ⇒ Object
32 33 34 |
# File 'lib/cloud_powers/synapse/pipe/stream.rb', line 32 def config { stream_name: @name, shard_count: @shard_count } end |
#create_resource ⇒ Object
15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
# File 'lib/cloud_powers/synapse/pipe/stream.rb', line 15 def create_resource begin @response = kinesis.create_stream(config) kinesis.wait_until(:stream_exists, stream_name: config[:stream_name]) @response.successful? # (http request successful && stream created)? rescue Aws::Kinesis::Errors::ResourceInUseException => e # TODO: write way more specs for these create methods. # all of this logic is suspicious. logger.info "#{name} already created" return if stream_status == 'ACTIVE' logger.info "Not ready for traffic..." sleep 1 @saved = true # acts like it would if it had to create the stream @linked = true end end |