lib/firehose/client/producer.rb
# Publish the message via HTTP.
Publish the message via HTTP.
Publish the message via HTTP .
def put(message, channel, opts, &block)
          ttl         = opts[:ttl]
          timeout     = opts[:timeout] || @timeout || DEFAULT_TIMEOUT
          buffer_size = opts[:buffer_size]
          deprecated  = opts[:deprecated]
          persist     = opts[:persist]
          response = conn.put do |req|
            req.options[:timeout] = timeout
            if conn.path_prefix.nil? || conn.path_prefix == '/'
              req.path = channel
            else
              if conn.path_prefix =~ /\/\Z/ || channel =~ /\A\//
                req.path = [conn.path_prefix, channel].compact.join
              else
                req.path = [conn.path_prefix, channel].compact.join('/')
              end
            end
            req.body = message
            req.headers['Cache-Control'] = "max-age=#{ttl.to_i}" if ttl
            req.headers["X-Firehose-Buffer-Size"] = buffer_size.to_s if buffer_size
            req.headers["X-Firehose-Deprecated"] = (!!deprecated).to_s if opts.include?(:deprecated)
            req.headers["X-Firehose-Persist"] = (!!persist).to_s if opts.include?(:persist)
          end
          response.on_complete do
            case response.status
            when 202 
              block.call(response) if block
            else
              response_data = response.inspect.gsub(/"Authorization"=>"Basic \S+"/, '"Authorization" => "Basic [HIDDEN]"')
              endpoint = "#{uri}/#{channel}".gsub(/:\/\/\S+@/, "://")
              error_handler.call PublishError.new("Could not publish #{message.inspect} to '#{endpoint}': #{response_data}")
            end
          end
          rescue Faraday::Error::TimeoutError, Faraday::ConnectionFailed => e
            error_handler.call TimeoutError.new(e)
        end
def put ( message , channel , opts , & block ) ttl = opts [ :ttl ] timeout = opts [ :timeout ] || @timeout || DEFAULT_TIMEOUT buffer_size = opts [ :buffer_size ] deprecated = opts [ :deprecated ] persist = opts [ :persist ] response = conn . put do | req | req . options [ :timeout ] = timeout if conn . path_prefix . nil? || conn . path_prefix == '/' req . path = channel else if conn . path_prefix =~ / \/ \Z / || channel =~ / \A \/ / req . path = [ conn . path_prefix , channel ] . compact . join else req . path = [ conn . path_prefix , channel ] . compact . join ( '/' ) end end req . body = message req . headers [ 'Cache-Control' ] = "max-age=#{ttl.to_i}" if ttl req . headers [ "X-Firehose-Buffer-Size" ] = buffer_size . to_s if buffer_size req . headers [ "X-Firehose-Deprecated" ] = ( ! ! deprecated ) . to_s if opts . include? ( :deprecated ) req . headers [ "X-Firehose-Persist" ] = ( ! ! persist ) . to_s if opts . include? ( :persist ) end response . on_complete do case response . status when 202 block . call ( response ) if block else response_data = response . inspect . gsub ( / "Authorization"=>"Basic  \S +" / , '"Authorization" => "Basic [HIDDEN]"' ) endpoint = "#{uri}/#{channel}" . gsub ( / : \/ \/ \S +@ / , "://" ) error_handler . call PublishError . new ( "Could not publish #{message.inspect} to '#{endpoint}': #{response_data}" ) end end rescue Faraday :: Error :: TimeoutError , Faraday :: ConnectionFailed => e error_handler . call TimeoutError . new ( e ) end
Publish the message via HTTP.
Publish the message via HTTP .
# This avoids a double / if the channel starts with a / too (which is expected). # Add a / so the prefix and channel aren't just rammed together. # Fire off the callback if everything worked out OK. # don't pass along basic auth header, if present # Hide Faraday with this Timeout exception, and through the error handler.
message
<not_specific>
channel
<not_specific>
opts
<not_specific>
block
<not_specific>