Re: [Cppcms-users] SSE keep-alive

[Christian G. <chr...@gm...>]

2013/4/2 Christian Gmeiner <chr...@gm...>:
> 2013/4/1 Artyom Beilis <art...@ya...>:
>>
>>
>> ----- Original Message -----
>>> From: Christian Gmeiner <chr...@gm...>
>>> To: cpp...@li...
>>> Cc:
>>> Sent: Monday, April 1, 2013 9:37 PM
>>> Subject: [Cppcms-users] SSE keep-alive
>>>
>>> HI all,
>>>
>>> I am trying to understand the keep-alive mechanism used in the SSE
>>> classes. I have the following problem:
>>>
>>> For test purposes I lowered the http timeout to 10 seconds, see
>>> keep-live is set to 1 second and the session timeout is set to 20
>>> seconds.
>>> Now a client opens the sse stream /sse/get and gets a "ping" message.
>>> Now in theory every second the see keep-alive worker should do its
>>> work, but it
>>> looks like long_pollers_ and streamers_ are empty. void
>>> event_source::keep_alive(char const *comment) gets called every
>>> seconds but as long_pollers_
>>> and streamers are empty no keep alive is send.
>>> There are two places where streamers_.insert is called: class
>>> post_send and void
>>> event_source::accept(booster::shared_ptr<cppcms::http::context> ctx).
>>>
>>> Maybe somebody can help me to under stand it!
>>>
>>> thanks
>>> --
>>> Christian Gmeiner, MSc
>>>
>>
>>
>>
>> I don't really understand your setup. Does a user connected to the
>> thread get the keep alive messages (empty comment messages like :keep-alive)
>> or not?
>>
>
> A user _NEVER_ gets the keep alive message.. thats why I am asking :)
>
> I have added some debug to void event_source::keep_alive(char const *comment):
> http://dpaste.com/hold/1043987/
>
>
> Here is an example log output:
> http://dpaste.com/hold/1044005/
>
>
> As you can see after then last data transfer via SSE the keep_alive
> gets called, but does nothing.
> After 5 seconds the connection run into a timeout.
>
>
> My used config:
>
> {
>         "service" : {
>                 "api" : "http",
>                 "ip" : "0.0.0.0",
>                 "port" : 8080
>         },
>         "http" : {
>                 "script" : "/tssw",
>                 "timeout" : 5,
>         },
>         "session" : {
>                 "expire" : "renew",
>                 "timeout" : 15,
>                 "location" : "server",
>                 "gc" : 10,
>                 "server": {
>                         "storage":"memory"
>                 }
>         },
>         "file_server" : {
>                 "enable" : true,
>                 "document_root" : "/opt/tssw/",
>                 "listing" : false,
>                 "alias" : [
>                         { "url" : "/assets" , "path" : "/opt/tssw/" },
>                         { "url" : "/downloads" , "path" : "/home/vis/" }
>                 ],
>         },
>         "security" : {
>                 "csrf" : {
>                         "enable" : true
>                 },
>                 "multipart_form_data_limit" : 1048576,
>                 "uploads_path" : "/home/vis/"
>         },
>         "localization" : {
>                 "messages" : {
>                         "paths" : [ "/opt/tssw/locale" ],
>                         "domains" : [ "tssw" ]
>                 },
>                 "locales" : [ "en.UTF-8", "de.UTF-8", "zh.UTF-8" ]
>         },
>         "logging" : {
>                 "level" : "debug",
>                 "syslog" : {
>                         "enable" : true,
>                         "id" : "tssw",
>                 },
>         }
> }
>
>
> 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.
>

Finally I got keep_alive working by changing on_send

    bool on_sent(event_stream &es)
    {
        bool something_send = false;
        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();
            something_send = true;
        }
        mutex.unlock();

        return something_send;
    }


Now I am looking to prevent expiration of the user session.

greets
--
Christian Gmeiner, MSc