Re: [Cppcms-users] SSE keep-alive

[Christian G. <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.

greets
--
Christian Gmeiner, MSc