Add support for Server-Sent Events

Author: kraih

Changes

@@ -1,5 +1,13 @@
 
-9.41  2025-05-13
+9.41  2025-07-03
+  - Added EXPERIMENTAL support for Server-Sent Events.
+  - Added EXPERIMENTAL module Mojo::SSE.
+  - Added EXPERIMENTAL sse attribute to Test::Mojo.
+  - Added EXPERIMENTAL get_sse_ok, post_sse_ok, sse_finish_ok, sse_finished_ok, sse_ok, sse_id_is, sse_id_isnt,
+    sse_text_is, sse_text_isnt, sse_text_like, sse_text_unlike, sse_type_is and sse_type_isnt methods to Test::Mojo.
+  - Added EXPERIMENTAL is_sse and write_sse methods to Mojo::Content.
+  - Added EXPERIMENTAL write_sse method to Mojolicious::Controller.
+  - Added EXPERIMENTAL sse event to Mojo::Content.
 
 9.40  2025-05-12
   - Added EXPERIMENTAL support for resumable file downloads.

README.md

@@ -30,8 +30,8 @@
   * A powerful **web development toolkit**, that you can use for all kinds of applications, independently of the web
     framework.
     * Full stack HTTP and WebSocket client/server implementation with IPv6, TLS, SNI, IDNA, HTTP/SOCKS5 proxy, UNIX
-      domain socket, Comet (long polling), Promises/A+, async/await, keep-alive, connection pooling, timeout, cookie,
-      multipart, and gzip compression support.
+      domain socket, Comet (long polling), Server-Sent Events (SSE), Promises/A+, async/await, keep-alive, connection
+      pooling, timeout, cookie, multipart, and gzip compression support.
     * Built-in non-blocking I/O web server, supporting multiple event loops as well as optional pre-forking and hot
       deployment, perfect for building highly scalable web services.
     * JSON and HTML/XML parser with CSS selector support.

examples/responses.pl

@@ -27,4 +27,10 @@
   die 'Hello World!';
 };
 
+get '/res6' => sub ($c) {
+  $c->write_sse({text => 'hello'});
+  $c->write_sse({text => 'world'});
+  $c->finish;
+};
+
 app->start;

lib/Mojo/Content.pm

@@ -4,6 +4,7 @@ use Mojo::Base 'Mojo::EventEmitter';
 use Carp                qw(croak);
 use Compress::Raw::Zlib qw(WANT_GZIP Z_STREAM_END);
 use Mojo::Headers;
+use Mojo::SSE    qw(build_event parse_event);
 use Scalar::Util qw(looks_like_number);
 
 has [qw(auto_decompress auto_relax relaxed skip_body)];
@@ -62,6 +63,8 @@ sub is_multipart {undef}
 
 sub is_parsing_body { (shift->{state} // '') eq 'body' }
 
+sub is_sse { (shift->headers->content_type // '') eq 'text/event-stream' }
+
 sub leftovers { shift->{buffer} }
 
 sub parse {
@@ -73,11 +76,14 @@ sub parse {
 
   # Chunked content
   $self->{real_size} //= 0;
-  if ($self->is_chunked && $self->{state} ne 'headers') {
+  if ($self->is_chunked) {
     $self->_parse_chunked;
     $self->{state} = 'finished' if ($self->{chunk_state} // '') eq 'finished';
   }
 
+  # SSE
+  elsif ($self->is_sse) { $self->_parse_sse }
+
   # Not chunked, pass through to second buffer
   else {
     $self->{real_size} += length $self->{pre_buffer};
@@ -158,6 +164,16 @@ sub write_chunk {
   return $self;
 }
 
+sub write_sse {
+  my ($self, $event, $cb) = @_;
+
+  $self->headers->content_type('text/event-stream') unless $self->{sse};
+  $self->{sse} = 1;
+
+  return $self->write unless defined $event;
+  return $self->write(build_event($event), $cb);
+}
+
 sub _build_chunk {
   my ($self, $chunk) = @_;
 
@@ -253,6 +269,20 @@ sub _parse_headers {
   $self->{header_size} = $self->{raw_size} - length $leftovers;
 }
 
+sub _parse_sse {
+  my $self = shift;
+
+  # Connection established
+  $self->emit('sse') unless $self->{sse};
+  $self->{sse} = 1;
+
+  # Parse SSE
+  while (my $event = parse_event(\$self->{pre_buffer})) { $self->emit(sse => $event) }
+
+  # Check buffer size
+  @$self{qw(state limit)} = ('finished', 1) if length($self->{pre_buffer} // '') > $self->max_buffer_size;
+}
+
 sub _parse_until_body {
   my ($self, $chunk) = @_;
 
@@ -319,6 +349,18 @@ Emitted when a new chunk of content arrives.
     say "Streaming: $bytes";
   });
 
+=head2 sse
+
+  $content->on(sse => sub ($content, $event) {...});
+
+Emitted when a new Server-Sent Event (SSE) connection has been established and for each new event that arrives. Note
+that this event is B<EXPERIMENTAL> and may change without warning!
+
+  $content->on(sse => sub ($content, $event) {
+    if ($event) { say "Type: $event->{type}, Data: $event->{data}" }
+    else        { say "SSE connection established" }
+  });
+
 =head1 ATTRIBUTES
 
 L<Mojo::Content> implements the following attributes.
@@ -480,6 +522,13 @@ False, this is not a L<Mojo::Content::MultiPart> object.
 
 Check if body parsing started yet.
 
+=head2 is_sse
+
+  my $bool = $content->is_sse;
+
+Check if C<Content-Type> header indicates Server-Sent Events (SSE). Note that this method is B<EXPERIMENTAL> and may
+change without warning!
+
 =head2 leftovers
 
   my $bytes = $content->leftovers;
@@ -541,6 +590,16 @@ dynamic content to be written later. You can write an empty chunk of data at any
     });
   });
 
+=head2 write_sse
+
+  $content = $content->write_sse;
+  $content = $content->write_sse($event);
+  $content = $content->write_sse($event => sub {...});
+
+Write Server-Sent Event (SSE) non-blocking, the optional drain callback will be executed once all data has been
+written. Calling this method without an event will finalize the response headers and allow for events to be written
+later. Note that this method is B<EXPERIMENTAL> and may change without warning!
+
 =head1 SEE ALSO
 
 L<Mojolicious>, L<Mojolicious::Guides>, L<https://mojolicious.org>.

lib/Mojo/SSE.pm

@@ -0,0 +1,84 @@
+package Mojo::SSE;
+use Mojo::Base -strict;
+
+use Carp       qw(croak);
+use Exporter   qw(import);
+use Mojo::Util qw(decode encode);
+
+our @EXPORT_OK = (qw(build_event parse_event));
+
+sub build_event {
+  my $event = shift;
+
+  my $data = $event->{text} // '';
+  croak 'Event data cannot contain newlines or linefeeds' if index($data, "\x0d") >= 0 || index($data, "\x0a") >= 0;
+
+  my @parts = defined $event->{type} ? ("event: $event->{type}") : ();
+  push @parts, "data: $data";
+  push @parts, "id: $event->{id}" if defined $event->{id};
+  return encode('UTF-8', join("\x0d\x0a", @parts, '', ''));
+}
+
+sub parse_event {
+  my $buffer = shift;
+  my $event  = {id => undef, type => 'message', text => ''};
+
+  while ($$buffer =~ s/^(.*?)(?:(?:\x0d\x0a|(?<!\x0d)\x0a|\x0d(?!\x0a)){2})//s) {
+
+    # Skip lines with encoding errors
+    next unless defined(my $lines = decode 'UTF-8', $1);
+
+    # Skip comments
+    next if $lines =~ /^\s*:/;
+
+    my $first = 0;
+    for my $line (split /(?:\x0d\x0a|(?<!\x0d)\x0a|\x0d(?!\x0a))/, $lines) {
+      if    ($line =~ /^event(?::\s*(\S.*))?$/) { $event->{type} = $1 // 'message' }
+      elsif ($line =~ /^data(?::\s*(.*))?$/)    { $event->{text} .= ($first++ ? "\n" : '') . ($1 // '') }
+      elsif ($line =~ /^id(?::\s*(.*))?$/)      { $event->{id} = $1 }
+    }
+
+    return $event;
+  }
+
+  return undef;
+}
+
+1;
+
+=encoding utf8
+
+=head1 NAME
+
+Mojo::SSE - Server-Sent Events
+
+=head1 SYNOPSIS
+
+  use Mojo::SSE qw(build_event parse_event);
+
+=head1 DESCRIPTION
+
+L<Mojo::SSE> implements the Server-Sent Events protocol. Note that this module is B<EXPERIMENTAL> and may change
+without warning!
+
+=head1 FUNCTIONS
+
+L<Mojo::SSE> implements the following functions, which can be imported individually.
+
+=head2 build_event
+
+  my $bytes = build_event $event, $chars;
+
+Build Server-Sent Event.
+
+=head2 parse_event
+
+  my $event = parse_event \$bytes;
+
+Parse Server-Sent Event. Returns C<undef> if no complete event was found.
+
+=head1 SEE ALSO
+
+L<Mojolicious>, L<Mojolicious::Guides>, L<https://mojolicious.org>.
+
+=cut

lib/Mojo/UserAgent/Transactor.pm

@@ -144,9 +144,11 @@ sub redirect {
     $headers->remove($_) for grep {/^content-/i} @{$headers->names};
   }
 
-  $new->res->content->auto_decompress(0) unless $self->compressed;
+  my $content = $new->res->content;
+  $content->auto_decompress(0) unless $self->compressed;
   my $headers = $new->req->url($location)->headers;
   $headers->remove($_) for qw(Authorization Cookie Host Referer);
+  if ($res->content->has_subscribers('sse')) { $content->on(sse => $_) for @{$res->content->subscribers('sse')} }
 
   return $new->previous($old);
 }

lib/Mojolicious/Controller.pm

@@ -329,6 +329,12 @@ sub write_chunk {
   return $self->rendered;
 }
 
+sub write_sse {
+  my ($self, $event, $cb) = @_;
+  $self->res->content->write_sse($event, $cb ? sub { shift; $self->$cb(@_) } : ());
+  return $self->rendered;
+}
+
 1;
 
 =encoding utf8
@@ -920,6 +926,21 @@ You can call L</"finish"> or write an empty chunk of data at any time to end the
   o!
   0
 
+=head2 write_sse
+
+  $c = $c->write_sse;
+  $c = $c->write_sse($event);
+  $c = $c->write_sse($event => sub ($c) {...});
+
+Write Server-Sent Event (SSE) non-blocking, the optional drain callback will be executed once all data has been
+written. Calling this method without an event will finalize the response headers and allow for events to be written
+later. Note that this method is B<EXPERIMENTAL> and might change without warning!
+
+  # Make sure previous event has been written before continuing (id and type are optional)
+  $c->write_sse({id => 23, type => 'greeting', text => 'Hello World!'} => sub ($c) {
+    $c->write_sse({id => 24, type => 'farewell', text => 'Goodbye World!'} => sub ($c) { $c->finish });
+  });
+
 =head1 HELPERS
 
 In addition to the L</"ATTRIBUTES"> and L</"METHODS"> above you can also call helpers provided by L</"app"> on

lib/Mojolicious/Guides/Cookbook.pod

@@ -998,12 +998,12 @@ result in much better performance, but also increases memory usage by up to 300K
 
 You can also use L<Mojo::Transaction::WebSocket/"with_protocols"> to negotiate a subprotocol.
 
-=head2 EventSource web service
+=head2 Server-Sent Events web service
 
-EventSource is a special form of long polling where you can use L<Mojolicious::Controller/"write"> to directly send DOM
-events from servers to clients. It is uni-directional, that means you will have to use Ajax requests for sending data
-from clients to servers, the advantage however is low infrastructure requirements, since it reuses the HTTP protocol
-for transport.
+Server-Sent Events (SSE) is a special form of long polling where you can use L<Mojolicious::Controller/"write_sse"> to
+directly send DOM events from servers to clients. It is uni-directional, that means you will have to use Ajax requests
+for sending data from clients to servers, the advantage however is low infrastructure requirements, since it reuses the
+HTTP protocol for transport.
 
   use Mojolicious::Lite -signatures;
 
@@ -1017,12 +1017,11 @@ for transport.
     $c->inactivity_timeout(300);
 
     # Change content type and finalize response headers
-    $c->res->headers->content_type('text/event-stream');
-    $c->write;
+    $c->write_sse;
 
     # Subscribe to "message" event and forward "log" events to browser
     my $cb = $c->app->log->on(message => sub ($log, $level, @lines) {
-      $c->write("event:log\ndata: [$level] @lines\n\n");
+      $c->write_sse({type => 'log', text => "[$level] @lines"});
     });
 
     # Unsubscribe from "message" event again once we are done

lib/Mojolicious/Types.pm

@@ -25,6 +25,7 @@ has mapping => sub {
     pdf      => ['application/pdf'],
     png      => ['image/png'],
     rss      => ['application/rss+xml'],
+    sse      => ['text/event-stream'],
     svg      => ['image/svg+xml'],
     ttf      => ['font/ttf'],
     txt      => ['text/plain;charset=UTF-8'],
@@ -114,6 +115,7 @@ L<Mojolicious::Types> manages MIME types for L<Mojolicious>.
   pdf      -> application/pdf
   png      -> image/png
   rss      -> application/rss+xml
+  sse      -> text/event-stream
   svg      -> image/svg+xml
   ttf      -> font/ttf
   txt      -> text/plain;charset=UTF-8