misc fixes to handling of "id" field; add last_event_id to StreamEvent (#26)

Author: eli-darkly

lib/ld-eventsource/client.rb

@@ -315,7 +315,7 @@ def read_stream(cxn)
           end
         end
       end
-      event_parser = Impl::EventParser.new(Impl::BufferedLineReader.lines_from(chunks))
+      event_parser = Impl::EventParser.new(Impl::BufferedLineReader.lines_from(chunks), @last_id)
 
       event_parser.items.each do |item|
         return if @stopped.value
@@ -331,7 +331,7 @@ def read_stream(cxn)
 
     def dispatch_event(event)
       @logger.debug { "Received event: #{event}" }
-      @last_id = event.id
+      @last_id = event.id if !event.id.nil?
 
       # Pass the event to the caller
       @on[:event].call(event)
@@ -354,7 +354,7 @@ def build_headers
         'Cache-Control' => 'no-cache',
         'User-Agent' => 'ruby-eventsource'
       }
-      h['Last-Event-Id'] = @last_id if !@last_id.nil?
+      h['Last-Event-Id'] = @last_id if !@last_id.nil? && @last_id != ""
       h.merge(@headers)
     end
   end

lib/ld-eventsource/events.rb

@@ -11,6 +11,10 @@ module SSE
   #     if there were multiple `data:` lines, they are concatenated with newlines
   # @!attribute id
   #   @return [String] the string that appeared after `id:` in the stream if any, or nil
+  # @!attribute last_event_id
+  #   @return [String] the `id:` value that was most recently seen in an event from
+  #     this stream; this differs from the `id` property in that it retains the same value
+  #     in subsequent events if they do not provide their own `id:`
   #
-  StreamEvent = Struct.new(:type, :data, :id)
+  StreamEvent = Struct.new(:type, :data, :id, :last_event_id)
 end

lib/ld-eventsource/impl/event_parser.rb

@@ -23,8 +23,9 @@ class EventParser
       # @param [Enumerator] lines  an enumerator that will yield one line of text at a time;
       #   the lines should not include line terminators
       #
-      def initialize(lines)
+      def initialize(lines, last_event_id = nil)
         @lines = lines
+        @last_event_id = last_event_id
         reset_buffers
       end
 
@@ -65,7 +66,10 @@ def process_field(name, value)
             @data << value
             @have_data = true
           when "id"
-            @id = value
+            if !value.include?("\x00")
+              @id = value
+              @last_event_id = value
+            end
           when "retry"
             if /^(?<num>\d+)$/ =~ value
               return SetRetryInterval.new(num.to_i)
@@ -76,7 +80,7 @@ def process_field(name, value)
 
       def maybe_create_event
         return nil if !@have_data
-        StreamEvent.new(@type || :message, @data, @id)
+        StreamEvent.new(@type || :message, @data, @id, @last_event_id)
       end
     end
   end

spec/client_spec.rb

@@ -7,19 +7,17 @@
 describe SSE::Client do
   subject { SSE::Client }
 
-  let(:simple_event_1) { SSE::StreamEvent.new(:go, "foo", "a")}
-  let(:simple_event_2) { SSE::StreamEvent.new(:stop, "bar", "b")}
+  let(:simple_event_1) { SSE::StreamEvent.new(:go, "foo")}
+  let(:simple_event_2) { SSE::StreamEvent.new(:stop, "bar")}
   let(:simple_event_1_text) { <<-EOT
 event: go
 data: foo
-id: a
 
 EOT
   }
   let(:simple_event_2_text) { <<-EOT
 event: stop
 data: bar
-id: b
 
 EOT
   }
@@ -254,8 +252,11 @@ def send_stream_content(res, content, keep_open:)
       server.setup_response("/") do |req,res|
         requests << req
         attempt += 1
-        send_stream_content(res, attempt == 1 ? simple_event_1_text : simple_event_2_text,
-          keep_open: attempt == 2)
+        if attempt == 1
+          send_stream_content(res, "data: foo\nid: a\n\n", keep_open: false)
+        else
+          send_stream_content(res, "data: bar\nid: b\n\n", keep_open: true)
+        end
       end
 
       event_sink = Queue.new
@@ -266,7 +267,7 @@ def send_stream_content(res, content, keep_open:)
       with_client(client) do |client|
         req1 = requests.pop
         req2 = requests.pop
-        expect(req2.header["last-event-id"]).to eq([ simple_event_1.id ])
+        expect(req2.header["last-event-id"]).to eq([ "a" ])
       end
     end
   end

spec/event_parser_spec.rb

@@ -9,6 +9,18 @@ def verify_parsed_events(lines:, expected_events:)
     expect(output).to eq(expected_events)
   end
 
+  it "parses an event with only data" do
+    lines = [
+      "data: def",
+      ""
+    ]
+    ep = subject.new(lines)
+    
+    expected_event = SSE::StreamEvent.new(:message, "def", nil)
+    output = ep.items.to_a
+    expect(output).to eq([ expected_event ])
+  end
+
   it "parses an event with all fields" do
     lines = [
       "event: abc",
@@ -18,23 +30,53 @@ def verify_parsed_events(lines:, expected_events:)
     ]
     ep = subject.new(lines)
 
-    expected_event = SSE::StreamEvent.new(:abc, "def", "1")
+    expected_event = SSE::StreamEvent.new(:abc, "def", "1", "1")
     output = ep.items.to_a
     expect(output).to eq([ expected_event ])
   end
 
-  it "parses an event with only data" do
+  it "retains last_event_id if latest event had no id" do
     lines = [
+      "event: abc",
       "data: def",
       ""
     ]
-    ep = subject.new(lines)
+    ep = subject.new(lines, "1")
 
-    expected_event = SSE::StreamEvent.new(:message, "def", nil)
+    expected_event = SSE::StreamEvent.new(:abc, "def", nil, "1")
     output = ep.items.to_a
     expect(output).to eq([ expected_event ])
   end
 
+  it "can override previous last_event_id with a blank id" do
+    lines = [
+      "event: abc",
+      "data: def",
+      "id:",
+      ""
+    ]
+    ep = subject.new(lines, "1")
+    
+    expected_event = SSE::StreamEvent.new(:abc, "def", "", "")
+    output = ep.items.to_a
+    expect(output).to eq([ expected_event ])
+  end
+
+  it "ignores id field if it contains a null character" do
+    # per SSE spec (9.2.6)
+    lines = [
+      "event: abc",
+      "data: def",
+      "id: 12\x0034",
+      ""
+    ]
+    ep = subject.new(lines, "1")
+    
+    expected_event = SSE::StreamEvent.new(:abc, "def", nil, "1")
+    output = ep.items.to_a
+    expect(output).to eq([ expected_event ])
+  end
+  
   it "parses an event with multi-line data" do
     lines = [
       "data: def",
@@ -146,8 +188,8 @@ def verify_parsed_events(lines:, expected_events:)
     ]
     ep = subject.new(lines)
 
-    expected_event_1 = SSE::StreamEvent.new(:abc, "def", "1")
-    expected_event_2 = SSE::StreamEvent.new(:message, "ghi", nil)
+    expected_event_1 = SSE::StreamEvent.new(:abc, "def", "1", "1")
+    expected_event_2 = SSE::StreamEvent.new(:message, "ghi", nil, "1")
     output = ep.items.to_a
     expect(output).to eq([ expected_event_1, expected_event_2 ])
   end