Reimplement the streaming chat-display layer

The message-delivery layer was gutted (user::msg_post commented out;
set_context/_send referenced only in comments, never defined), so the
chat could log in but never show or send messages: the stream frame
returned an empty response (502 behind Traefik) and posted messages went
nowhere.

Reimplemented server-streamed chat:
- user: add i_stream_fd (long-lived chat-display connection) + b_stream_ready
  + s_msg buffer. msg_post appends to s_msg and flushes to the fd once the
  initial response is sent (best-effort non-blocking write; buffer on
  EAGAIN; on hard write error drop the reference and mark offline).
  flush_stream/clear_stream helpers; clear_stream on clean()/disconnect.
- reqp (event=stream): build the initial HTTP response (headers, no
  Content-Length so it streams incrementally, + the parsed stream.html
  body), attach this fd to the user, set p_context->p_user, KEEP_ALIVE=yes.
- sock: route KEEP_ALIVE responses through handle_stream_write (sends the
  initial page, then arms a persistent read event) and handle_stream_read
  (drains stray bytes; on EOF/error reaps the context and clears the
  user's stream). The connection stays open for pushed messages.
- context: init p_user=NULL; track i_buf_len across reads (POST body behind
  a proxy arrives in a separate segment).

Verified locally: login -> open stream -> post message -> message appears
in the stream; two users in a room both receive a posted message.

6 files changed, 185 insertions, 22 deletions

diff --git a/ychat/src/chat/user.cpp b/ychat/src/chat/user.cpp
index 80f4c65..1308cb0 100644
--- a/ychat/src/chat/user.cpp
+++ b/ychat/src/chat/user.cpp
@@ -32,6 +32,8 @@
 using namespace std;
 
 #include <errno.h>
+#include <unistd.h>
+#include <cstring>
 
 user::user() : name( "" )
 {
@@ -44,7 +46,9 @@ user::user( string s_name ) : name( s_name )
 }
 
 user::~user()
-{}
+{
+  // The stream fd is owned by its context, not by the user.
+}
 
 void
 user::initialize()
@@ -56,6 +60,8 @@ user::initialize()
 
   this -> l_messages_recv = 0;
   this -> p_sock = NULL;
+  this -> i_stream_fd = -1;
+  this -> b_stream_ready = false;
   this -> p_room = NULL;
   this -> s_msg = "";
   this -> b_is_reg = false;
@@ -78,6 +84,7 @@ user::clean()
   set_invisible(false);
   set_away(false, "");
   set_sock(NULL);
+  clear_stream();
 }
 
 void
@@ -441,25 +448,71 @@ user::set_sock(_socket* p_sock)
 }
 
 void
+user::set_stream_fd(int i_fd)
+{
+  // The context owns the fd and closes it on disconnect; the user only
+  // stores the number so msg_post can write to it. -1 = no stream.
+  i_stream_fd = i_fd;
+}
+
+int
+user::get_stream_fd()
+{
+  return i_stream_fd;
+}
+
+// Write a chat message to this user's open stream connection.
+// Single-threaded (libevent loop), so no locking needed. Messages are
+// buffered in s_msg until the initial HTTP response has been sent
+// (b_stream_ready), then flushed. Best-effort non-blocking write; anything
+// that can't be written now stays buffered for the next flush. On a hard
+// write error the peer is gone: drop our reference (the stream context's
+// read event will reap it and close the fd) and mark the user offline.
+void
 user::msg_post( string *p_msg )
 {
-  /*
   ++l_messages_recv;
-  if (p_sock == NULL) {
+
+  if ( i_stream_fd < 0 )
+    return; // no open stream connection; nothing to deliver to
+
   s_msg.append(*p_msg);
-  return;
+  if ( ! b_stream_ready )
+    return; // initial response not yet sent; keep buffered
 
-  }
-  else if (!s_msg.empty())
+  flush_stream();
+}
+
+void
+user::flush_stream()
+{
+  while ( ! s_msg.empty() && i_stream_fd >= 0 )
   {
-  wrap::SOCK->_send(p_sock, s_msg.c_str(), s_msg.size() );
-  s_msg.clear();
+    ssize_t n = write( i_stream_fd, s_msg.data(), s_msg.size() );
+    if ( n > 0 )
+    {
+      s_msg.erase( 0, n );
+      continue;
+    }
+    if ( n == 0 )
+      break;
+
+    if ( errno == EAGAIN || errno == EINTR )
+      break; // socket buffer full; keep buffered for later
+
+    // Hard error (EPIPE, ECONNRESET, ...): peer gone.
+    clear_stream();
+    set_online( false );
+    return;
   }
+}
 
-  if ( 0 > wrap::SOCK->_send(p_sock, p_msg->c_str(), p_msg->size() ) ) {
-  cout << "psock: " << (int) p_sock << endl;
-  set_online( false );
-  */
+void
+user::clear_stream()
+{
+  i_stream_fd = -1;
+  b_stream_ready = false;
+  s_msg.clear();
 }
 /*
     void
diff --git a/ychat/src/chat/user.h b/ychat/src/chat/user.h
index 1369067..d8ea777 100644
--- a/ychat/src/chat/user.h
+++ b/ychat/src/chat/user.h
@@ -48,6 +48,8 @@ private:
 
   // private members:
   _socket *p_sock;
+  int   i_stream_fd; // long-lived stream connection fd (-1 = none)
+  bool  b_stream_ready; // initial HTTP response sent; ok to push messages
   string s_msg;
   bool b_online; // true if user is online.
   bool b_has_sess; // true if user already has a session!
@@ -111,6 +113,12 @@ public:
   bool get_is_gag();
   void set_online( bool b_online );
   void set_sock(_socket *p_sock);
+  // Stream (long-lived chat-display) connection for this user.
+  void set_stream_fd(int i_fd);
+  int  get_stream_fd();
+  void set_stream_ready(bool b) { b_stream_ready = b; }
+  void flush_stream();   // write buffered s_msg to the stream fd
+  void clear_stream();   // drop stream state (disconnect/logout)
   void set_fake( bool b_fake );
   void set_invisible( bool b_invisible );
   void set_has_sess( bool b_has_sess );
diff --git a/ychat/src/reqp.cpp b/ychat/src/reqp.cpp
index 74b2110..b68bed7 100644
--- a/ychat/src/reqp.cpp
+++ b/ychat/src/reqp.cpp
@@ -104,16 +104,24 @@ reqp::parse(context *p_context)
         // if a chat stream
         else if ( s_event == "stream" )
         {
-          /*
-          string s_resp(s_http+s_http_stream);
-          s_resp.append(s_http_colength + tool::int2string(s_response.size()) + "\r\n" +
-               s_http_cotype + map_params["content-type"]
-                // + s_http_cotype_add
-          	 + "\r\n\r\n" );*/
-          /*		  string s_resp("");
-                    s_resp.append(wrap::HTML->parse(map_params));
-          		  p_user->set_context(p_context);
-          		  */
+          // Long-lived streaming chat-display connection. Build the initial
+          // HTTP response (headers + the stream.html template) into p_response;
+          // handle_client_read sends it via the keep-alive path and keeps the
+          // connection open. Attach this fd to the user so msg_post can push
+          // subsequent chat messages to it.
+          string s_body = wrap::HTML->parse( map_params );
+
+          string s_resp;
+          s_resp.append( s_http ); // HTTP/1.1 200 OK ... Cache-Control ... Connection: close
+          s_resp.append( s_http_cotype + map_params["content-type"] + s_http_cotype_add + "\r\n" );
+          // No Content-Length: the body streams incrementally until close.
+          s_resp.append( "\r\n" );
+          s_resp.append( s_body );
+
+          *p_response = s_resp;
+
+          p_user->set_stream_fd( p_context->i_fd );
+          p_context->p_user = p_user;
           map_params["KEEP_ALIVE"] = "yes";
           return;
         }
diff --git a/ychat/src/sock/context.cpp b/ychat/src/sock/context.cpp
index ac25e95..0e57fed 100644
--- a/ychat/src/sock/context.cpp
+++ b/ychat/src/sock/context.cpp
@@ -37,6 +37,7 @@ context::context(sock *p_sock, struct event *p_event, int i_fd)
   this->i_fd = i_fd;
   this->i_buf_len = 0;
   this->c_buf[0] = '\0';
+  this->p_user = NULL;
 }
 
 context::~context()
diff --git a/ychat/src/sock/sock.cpp b/ychat/src/sock/sock.cpp
index 070d540..327b5ef 100644
--- a/ychat/src/sock/sock.cpp
+++ b/ychat/src/sock/sock.cpp
@@ -493,6 +493,17 @@ sock::handle_client_read(int i_fd, short event, void *p_arg)
   p_context->p_response = new string("");
   p_sock->get_req_parser()->parse(p_context);
 
+  if ( (*p_context->p_map_params)["KEEP_ALIVE"] == "yes" )
+  {
+    // Streaming chat-display connection: send the initial response, then
+    // keep the fd open for pushed messages and watch for disconnect.
+    struct event *p_ev = new struct event;
+    p_context->p_event = p_ev;
+    event_set( p_ev, i_fd, EV_WRITE, handle_stream_write, p_context );
+    event_add( p_ev, NULL );
+    return;
+  }
+
   struct event *p_ev_handle_client_write = new struct event;
   p_context->p_event = p_ev_handle_client_write;
 
@@ -500,6 +511,86 @@ sock::handle_client_read(int i_fd, short event, void *p_arg)
   event_add(p_ev_handle_client_write, NULL);
 }
 
+// Finish sending the initial stream response, then switch to a persistent
+// read event that detects the client disconnecting. Does NOT delete the
+// context (the connection stays open for streamed chat messages).
+void
+sock::handle_stream_write(int i_fd, short event, void *p_arg)
+{
+  context *p_context = static_cast<context*>(p_arg);
+  string *p_response = p_context->p_response;
+
+  if ( p_response && ! p_response->empty() )
+  {
+    ssize_t n = write( i_fd, p_response->data(), p_response->size() );
+    if ( n > 0 )
+      p_response->erase( 0, n );
+    else if ( n < 0 && ( errno == EAGAIN || errno == EINTR ) )
+    {
+      event_add( p_context->p_event, NULL );
+      return;
+    }
+    else if ( n < 0 )
+    {
+      // Peer gone before the initial response was sent.
+      if ( p_context->p_user )
+        p_context->p_user->clear_stream();
+      delete p_context;
+      return;
+    }
+
+    if ( ! p_response->empty() )
+    {
+      event_add( p_context->p_event, NULL );
+      return;
+    }
+  }
+
+  // Initial response fully sent: free it, mark the user's stream as ready
+  // to receive pushed messages, flush anything already buffered, then arm
+  // a persistent read event to detect disconnect.
+  delete p_response;
+  p_context->p_response = NULL;
+
+  if ( p_context->p_user )
+  {
+    p_context->p_user->set_stream_ready( true );
+    p_context->p_user->flush_stream();
+  }
+
+  event_set( p_context->p_event, i_fd, EV_READ | EV_PERSIST, handle_stream_read, p_context );
+  event_add( p_context->p_event, NULL );
+}
+
+// The stream connection is quiet except for disconnect. Drain any stray
+// bytes the client sends; on EOF/error reap the context and clear the
+// user's stream reference.
+void
+sock::handle_stream_read(int i_fd, short event, void *p_arg)
+{
+  context *p_context = static_cast<context*>(p_arg);
+  char c_buf[256];
+
+  for (;;)
+  {
+    ssize_t n = read( i_fd, c_buf, sizeof(c_buf) );
+    if ( n > 0 )
+      continue; // drain unexpected client data
+    if ( n == 0 )
+      break;    // EOF -> disconnect
+    // n < 0
+    if ( errno == EAGAIN || errno == EINTR )
+      return;   // still connected; persistent event stays armed
+    break;      // error -> disconnect
+  }
+
+  if ( p_context->p_event )
+    event_del( p_context->p_event );
+  if ( p_context->p_user )
+    p_context->p_user->clear_stream();
+  delete p_context;
+}
+
 void
 sock::handle_client_write(int i_fd, short event, void *p_arg)
 {
diff --git a/ychat/src/sock/sock.h b/ychat/src/sock/sock.h
index b09afb1..b22cd20 100644
--- a/ychat/src/sock/sock.h
+++ b/ychat/src/sock/sock.h
@@ -74,6 +74,8 @@ protected:
   static void handle_client_read(int i_fd, short event, void *p_arg);
   static void handle_client_write(int i_fd, short event, void *p_arg);
   static void handle_client_stream_write(int i_fd, short event, void *p_arg);
+  static void handle_stream_write(int i_fd, short event, void *p_arg);
+  static void handle_stream_read(int i_fd, short event, void *p_arg);
 
   int set_nonblock(int i_sock);