class OpenUI::Rails::Stream::Accumulator

Client-side (Ruby) sequence accumulator mirroring what the browser must do: apply ordered deltas, ignore duplicates, and treat a missing sequence as “fetch the authoritative snapshot” rather than executing a malformed partial. Used by server tests and by hosts that consume their own stream.

Attributes

message_id[R]

source[R]

Public Class Methods

new(message_id:) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 14
def initialize(message_id:)
  @message_id = message_id
  @source = +""
  @next_sequence = 0
  @completed = false
  @error = nil
  @needs_snapshot = false
end

Public Instance Methods

apply(envelope) click to toggle source

@return [Symbol] one of :applied, :duplicate, :snapshot, :completed,

:error, :ignored

# File lib/openui/rails/stream/accumulator.rb, line 25
def apply(envelope)
  return :ignored unless envelope.message_id == message_id
  unless envelope.snapshot?
    return :duplicate if envelope.sequence < @next_sequence
    return :ignored if @completed || error?
    return :gap if @needs_snapshot
    if envelope.sequence > @next_sequence
      @needs_snapshot = true
      return :gap
    end
  end

  case envelope.type
  when Envelope::DELTA then apply_delta(envelope)
  when Envelope::SNAPSHOT then apply_snapshot(envelope)
  when Envelope::COMPLETED then apply_completed(envelope)
  when Envelope::ERROR then apply_error(envelope)
  else :ignored
  end
end

apply_completed(envelope) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 72
def apply_completed(envelope)
  @completed = true
  @next_sequence = [ @next_sequence, envelope.sequence + 1 ].max
  :completed
end

apply_delta(envelope) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 54
def apply_delta(envelope)
  @source << envelope.delta
  @next_sequence += 1
  :applied
end

apply_error(envelope) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 78
def apply_error(envelope)
  @error = envelope
  @next_sequence += 1
  :error
end

apply_snapshot(envelope) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 60
def apply_snapshot(envelope)
  minimum = @needs_snapshot ? @next_sequence - 1 : @next_sequence
  return :duplicate if envelope.sequence < minimum

  @source = +envelope.response.dup
  @next_sequence = envelope.sequence + 1
  @needs_snapshot = false
  @completed = !envelope.streaming?
  @error = nil
  :snapshot
end

completed?(= @completed) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 47
        def completed? = @completed
        def error? = !@error.nil?
        def needs_snapshot? = @needs_snapshot
        def error = @error

        private

        def apply_delta(envelope)
          @source << envelope.delta
          @next_sequence += 1
          :applied
        end

        def apply_snapshot(envelope)
          minimum = @needs_snapshot ? @next_sequence - 1 : @next_sequence
          return :duplicate if envelope.sequence < minimum

          @source = +envelope.response.dup
          @next_sequence = envelope.sequence + 1
          @needs_snapshot = false
          @completed = !envelope.streaming?
          @error = nil
          :snapshot
        end

        def apply_completed(envelope)
          @completed = true
          @next_sequence = [ @next_sequence, envelope.sequence + 1 ].max
          :completed
        end

        def apply_error(envelope)
          @error = envelope
          @next_sequence += 1
          :error
        end
      end
    end
  end
end

error(= @error) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 50
  def error = @error

  private

  def apply_delta(envelope)
    @source << envelope.delta
    @next_sequence += 1
    :applied
  end

  def apply_snapshot(envelope)
    minimum = @needs_snapshot ? @next_sequence - 1 : @next_sequence
    return :duplicate if envelope.sequence < minimum

    @source = +envelope.response.dup
    @next_sequence = envelope.sequence + 1
    @needs_snapshot = false
    @completed = !envelope.streaming?
    @error = nil
    :snapshot
  end

  def apply_completed(envelope)
    @completed = true
    @next_sequence = [ @next_sequence, envelope.sequence + 1 ].max
    :completed
  end

  def apply_error(envelope)
    @error = envelope
    @next_sequence += 1
    :error
  end
end

error?(= !@error.nil?) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 48
      def error? = !@error.nil?
      def needs_snapshot? = @needs_snapshot
      def error = @error

      private

      def apply_delta(envelope)
        @source << envelope.delta
        @next_sequence += 1
        :applied
      end

      def apply_snapshot(envelope)
        minimum = @needs_snapshot ? @next_sequence - 1 : @next_sequence
        return :duplicate if envelope.sequence < minimum

        @source = +envelope.response.dup
        @next_sequence = envelope.sequence + 1
        @needs_snapshot = false
        @completed = !envelope.streaming?
        @error = nil
        :snapshot
      end

      def apply_completed(envelope)
        @completed = true
        @next_sequence = [ @next_sequence, envelope.sequence + 1 ].max
        :completed
      end

      def apply_error(envelope)
        @error = envelope
        @next_sequence += 1
        :error
      end
    end
  end
end

needs_snapshot?(= @needs_snapshot) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 49
    def needs_snapshot? = @needs_snapshot
    def error = @error

    private

    def apply_delta(envelope)
      @source << envelope.delta
      @next_sequence += 1
      :applied
    end

    def apply_snapshot(envelope)
      minimum = @needs_snapshot ? @next_sequence - 1 : @next_sequence
      return :duplicate if envelope.sequence < minimum

      @source = +envelope.response.dup
      @next_sequence = envelope.sequence + 1
      @needs_snapshot = false
      @completed = !envelope.streaming?
      @error = nil
      :snapshot
    end

    def apply_completed(envelope)
      @completed = true
      @next_sequence = [ @next_sequence, envelope.sequence + 1 ].max
      :completed
    end

    def apply_error(envelope)
      @error = envelope
      @next_sequence += 1
      :error
    end
  end
end

next_sequence(= @next_sequence) click to toggle source

# File lib/openui/rails/stream/accumulator.rb, line 46
      def next_sequence = @next_sequence
      def completed? = @completed
      def error? = !@error.nil?
      def needs_snapshot? = @needs_snapshot
      def error = @error

      private

      def apply_delta(envelope)
        @source << envelope.delta
        @next_sequence += 1
        :applied
      end

      def apply_snapshot(envelope)
        minimum = @needs_snapshot ? @next_sequence - 1 : @next_sequence
        return :duplicate if envelope.sequence < minimum

        @source = +envelope.response.dup
        @next_sequence = envelope.sequence + 1
        @needs_snapshot = false
        @completed = !envelope.streaming?
        @error = nil
        :snapshot
      end

      def apply_completed(envelope)
        @completed = true
        @next_sequence = [ @next_sequence, envelope.sequence + 1 ].max
        :completed
      end

      def apply_error(envelope)
        @error = envelope
        @next_sequence += 1
        :error
      end
    end
  end
end