Sha256: da4342d2ac84faaf4cdcffbcac268a220c6aa342a0fb09cf954db7b25a958ef8

Contents?: true

Size: 997 Bytes

Versions: 8

Compression:

Stored size: 997 Bytes

Contents

module Datadog
  # Pipeline
  module Pipeline
    require_relative 'pipeline/span_filter'
    require_relative 'pipeline/span_processor'

    @mutex = Mutex.new
    @processors = []

    def self.before_flush(*processors, &processor_block)
      processors = [processor_block] if processors.empty?

      @mutex.synchronize do
        @processors.concat(processors)
      end
    end

    def self.process!(traces)
      @mutex.synchronize do
        traces
          .map(&method(:apply_processors!))
          .select(&:any?)
      end
    end

    def self.processors=(value)
      @processors = value
    end

    def self.apply_processors!(trace)
      result = @processors.inject(trace) do |current_trace, processor|
        processor.call(current_trace)
      end

      result || []
    rescue => e
      Datadog::Logger.log.debug(
        "trace dropped entirely due to `Pipeline.before_flush` error: #{e}"
      )

      []
    end

    private_class_method :apply_processors!
  end
end

Version data entries

8 entries across 8 versions & 1 rubygems

Version Path
ddtrace-0.34.2 lib/ddtrace/pipeline.rb
ddtrace-0.34.1 lib/ddtrace/pipeline.rb
ddtrace-0.34.0 lib/ddtrace/pipeline.rb
ddtrace-0.33.1 lib/ddtrace/pipeline.rb
ddtrace-0.33.0 lib/ddtrace/pipeline.rb
ddtrace-0.32.0 lib/ddtrace/pipeline.rb
ddtrace-0.31.1 lib/ddtrace/pipeline.rb
ddtrace-0.31.0 lib/ddtrace/pipeline.rb