Re: [Cppcms-users] "endless" stream of data

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

>________________________________
> From: Marcel Hellwig <ke...@co...>
>To: Artyom Beilis <art...@ya...>; cpp...@li... 
>Sent: Tuesday, December 27, 2011 9:34 PM
>Subject: Re: [Cppcms-users] "endless" stream of data
> 
>Alright, my code is a bit different from yours and from the chat example but else it doesn't work.
>
>void SSE::loop(std::string /*page*/)
>{
>    response().io_mode(cppcms::http::response::asynchronous);

Don't need this... As if the application is mounted as asynchronous
(make sure you did) then it is the default mode.

>    response().set_header("Content-Type", "text/event-stream");
>
>    int counter = (rand() % 10) + 1;

General notes below

You write the code as it is synchronous code, it is not!
You don't work like this.


>
>    while (true) {
>        response().out() << "event: ping\n"

><< "data: {\"time\": \""
><< "foo" //note: this is just an example
><< "\"}\n\n";
>        counter--;
>
>        if (!counter) {
>            response().out() << "data: is a message at time "
><< "bar" << "\n\n";
>            counter = (rand() % 10) + 1;
>        }
>
>        context().async_flush_output(boost::bind(&SSE::handler, this, -1));

You should use a handler per call, Once you called async flush
you should not do further operation till you receive completion
notifications

>        booster::ptime::sleep(booster::ptime::seconds(1));

You block there entire event loop, you need to use booster::aio::deadline_timer
and handle events asynchronously...

>
>    }
>}
>
>void SSE::handler(int status)
>{
>    std::cout << "Status: " << status << std::endl;
>}
>
>two major questions. You told me to use release_context but if I do so, the connection "aborts" (access to unassigned context) while the second loop on response().out()

You release it and you become responsible for the context life time,
you need to keep it withing your class as long as you need.

>How can I figure out when the user "leaves" the page, so I can stop producing tons of messages?
>

Two ways:

a) The completion handler of async_flush_output returns a error status
b) You can using async_on_peer_reset callback.

Take a look in the chat example in the cppcms sources

get(std::string no) {
...

        else if(pos == messages_.size()) {
            // Get the context <-------------------------------------------
            booster::shared_ptr<cppcms::http::context> context=release_context();
            // Keep it         <--------------------------------------------
            waiters_.insert(context);
            // Handle disconnect Asynchronously!!!
            context->async_on_peer_reset(
                bind(
                    &chat::remove_context,
                    booster::intrusive_ptr<chat>(this),
                    context));
        }
...
}

Now take a look on function broadcast:

   void broadcast()
    {
        for(waiters_type::iterator waiter=waiters_.begin();waiter!=waiters_.end();++waiter) {
            (*waiter)->response().set_plain_text_header();
            (*waiter)->response().out() << messages_.back();
            // n
            (*waiter)->async_complete_response();
        }
        waiters_.clear();
    }

Lets change the line

            (*waiter)->async_complete_response();

To
            (*waiter)->async_flush_output(
                     boost::bind(&chat::new_handle,
                                  booster::intrusive_ptr<chat>(this),
                                  _1,
                                  *waiter,
                                   messages_.size()));

And add new handler for "endless handling":

   void new_handle(cppcms::http::context::completion_type t,
                   booster::shared_ptr<cppcms::http::context> context,
                   size_t broadcasted)
   {
          if(t) return; // non-zero error
          if(broadcasted == messages_.size()) 
              // no new messages mark for waiting once agains
              waiters_.push_back(context);
          else {
              // send all we got meanwhile
              while(broadcasted < messages_.size()) {
                     context->response().out() << messages_[broadcaster] <<'\n;
                     broadcasted++;
              }
              // send message asychronously
              context->async_flush_output(
                     boost::bind(
                          &chat::new_handle,
                           booster::intrusive_ptr<chat>(this),
                           _1,
                           context,
                           messages_.size()));
          }
   }



And you got chat with "endless" stream, of course it would not work with
the given XMLHttpRequest example but you get the general idea.

BOTTOM LINE
-----------

Think asynchronously... 

Regards,
  Artyom