Re: [Cppcms-users] SSE keep-alive

[Artyom B. <art...@ya...>]

Your code is incorrect, you must not use Mutexes...

Instead of this:

>     void push(std::string const &event,std::string const &data,bool 
> send=true)
>     {
>         message msg;
>         msg.event = event;
>         msg.data = data;
> 
>         mutex.lock();
>         messages_.push(msg);
>         mutex.unlock();
> 
>         if(send)
>             broadcast();
>     }


That I assume is called form an external thread you need to do something different


void thread_safe_push(std::string const &event,std::string const &data)
{      message msg;
      msg.event = event;
      msg.data = data;


      service().post([=] {  (lambda expression)

         // EXECUTED IN THE EVENT LOOP THREAD!!!
         messages_.push(msg);
         broadcast();

      }); 

}


or without C++11

void thread_safe_push(std::string const &event,std::string const &data)
{      message msg;
      msg.event = event;
      msg.data = data;


      service().post(boost::bind(&event_fifo::thread_unsef_push,this,msg));

}


void thread_unsefe_push(message const &msg)

{      messages_.push(msg);
      broadcast();

}


----- Original Message -----
> From: Christian Gmeiner <chr...@gm...>
> }
> 
> 
> My used SSE application looks like:
> 
> SSE::SSE(cppcms::service &srv) : cppcms::application(srv)
> {
>     stream_ = sse::event_fifo::create(srv.get_io_service());
>     stream_->enable_keep_alive(1);
> 
>     dispatcher().assign("/get",&SSE::get,this);
> }
> 
> void SSE::get()
> {
>     stream_->accept(release_context());
> }
> 
> void SSE::enqueue(std::string const &event, std::string const &data)
> {
>     stream_->push(event, data);
> }
> 
> 
> And my used event_fifo class:
> 
> class event_fifo : public event_source {
> protected:
>     event_fifo(booster::aio::io_service &srv) :
>         event_source(srv)
>     {
>     }
> public:
>     ///
>     /// Create a queue of maximal size \a size, such that user that
> connects too late
>     /// it would be able to receive at most \a size latest messages
>     ///
>     static booster::shared_ptr<event_fifo> create(
> booster::aio::io_service &srv)
>     {
>         booster::shared_ptr<event_fifo> p(new event_fifo(srv));
>         return p;
>     }
> 
>     virtual void accept(booster::shared_ptr<cppcms::http::context> ctx)
>     {
>         // remove old messages
>         mutex.lock();
>         while (messages_.empty() == false)
>         {
>             messages_.pop();
>         }
>         mutex.unlock();
> 
>         event_source::accept(ctx);
> 
>         // we need to send something that the EventSource on the client side
>         // gets notified that the stream is open.
>         push("welcome", "welcome");
>     }
> 
>     ///
>     /// put a message into the fifo
>     ///
>     /// If \a send is false the messages are not dispatched
>     /// immediately, you can dispatch them later by calling
> broadcast() or by calling push
>     /// event with send=true
>     ///
>     void push(std::string const &data,bool send=true)
>     {
>         push(std::string(),data,send);
>     }
> 
>     ///
>     /// put a message into the fifo
>     ///
>     /// If \a send is false the messages are not dispatched
>     /// immediately, you can dispatch them later by calling
> broadcast() or by calling push
>     /// event with send=true
>     ///
>     void push(std::string const &event,std::string const &data,bool 
> send=true)
>     {
>         message msg;
>         msg.event = event;
>         msg.data = data;
> 
>         mutex.lock();
>         messages_.push(msg);
>         mutex.unlock();
> 
>         if(send)
>             broadcast();
>     }
> 
> protected:
>     bool on_sent(event_stream &es)
>     {
>         size_t last_id = es.last_integer_id();
>         size_t id = last_id;
> 
>         mutex.lock();
>         while (messages_.empty() == false)
>         {
>             id++;
>             message &msg = messages_.front();
>             es.write(msg.data, id, msg.event);
>             messages_.pop();
>         }
>         mutex.unlock();
> 
>         return true;
>     }
> 
> private:
>     struct message {
>         std::string event;
>         std::string data;
>     };
>     std::queue<message> messages_;
>     booster::mutex mutex;
> };
> 
> } // namespace sse
> 
> Oh... it is planed to have one SSE stream per session.
> If you need it I can provide you a simple demo application.
> 
> greets
> --
> Christian Gmeiner, MSc
>



 
Artyom Beilis
--------------
CppCMS - C++ Web Framework:   http://cppcms.com/
CppDB - C++ SQL Connectivity: http://cppcms.com/sql/cppdb/