Introducción

En Sinatra 1.3 se introdujo el stream helper. El stream object imita a un objeto IO.

1. Código del método stream
2. Class Stream code: Class of the response body in case you use #stream

En ocasiones queremos empezar a enviar datos mientras se esta generando aún el cuerpo de la respuesta. Puede que incluso queramos mantener el envío hasta que el cliente cierra la conexión.

Para eso podemos usar el stream helper:

[~/sinatra/sinatra-streaming/intro-streaming(master)]$ cat legendary.rb 
require 'sinatra'

before do
 content_type 'text/plain'
end

get '/' do
  stream do |out|
    out << "It's gonna be legen -\n"
    sleep 0.5
    out << " (wait for it) \n"
    sleep 1
    out << "- dary!\n"
  end
end

El objeto out que el método stream pasa al bloque es un objeto de la clase Sinatra::Helpers::Stream.

Este objeto representa el cuerpo de la respuesta en el caso en que se use stream.

El método stream permite enviar datos al cliente incluso cuando el cuerpo del cliente no ha sido generado completamente.

Utilidades del Streaming

Esto nos permite implementar

• streaming APIs,
• Server Sent Events. Server-sent events es una tecnología que proporciona notificaciones push desde el servidor a un navegador cliente en forma de eventos DOM
• y que puede ser usada como base para WebSockets. WebSocket es una tecnología que proporciona un canal de comunicación bidireccional y full-duplex sobre un único socket TCP. Está diseñada para ser implementada en navegadores y servidores web, pero puede utilizarse por cualquier aplicación cliente/servidor.
• Puede también ser utilizada para incrementar el throughput si parte pero no todo el contenido depende de un recurso que es lento.

Dependencia del Servidor

Nótese que la conducta de streaming, en particular el número de solicitudes concurrentes, depende en gran medida del servidor web que sirve la aplicación.

• Algunos servidores como Webrick no soportan streaming. En tal caso el cuerpo es envíado en una tacada.
• Shotgun tampoco soporta streaming

Sinatra::Streaming

1. Sinatra 1.3 introduced the stream helper.
2. The addon provided by the gem sinatra-contrib improves the streaming API by making the stream object immitate an IO object, making the body play nicer with middleware unaware of streaming.

This is useful when passing the stream object to a library expecting an IO or StringIO object.

06:31][~/srcSTW/streaming/upandrunning_streaming]$ cat -n simple.rb 
     1  # http://www.sinatrarb.com/contrib/streaming.html
     2  # $ gem install sinatra-contrib
     3  require 'sinatra'
     4  require 'sinatra/streaming'
     5  set server: 'thin'
     6  #set server: 'unicorn'
     7  
     8  get '/' do
     9    stream do |out|
    10      puts out.methods
    11      out.puts "Hello World!", "How are you?"
    12      out.write "Written #{out.pos} bytes so far!\n"
    13      out.putc(65) unless out.closed?
    14      out.flush
    15    end
    16  end

out es un objeto Sinatra::Helpers::Stream.

sinatra/streaming en Aplicaciones Modulares

Use the top-level helpers method to introduce the methods in the module Sinatra::Streaming for use in route handlers and templates:

require "sinatra/base"
require "sinatra/streaming"

class MyApp < Sinatra::Base
  helpers Sinatra::Streaming
end

Manteniendo la Conexión Abierta: keep_open

• Si en el método stream el parámetro opcional keep_open se pone a true close no será llamado al finalizar el bloque, permitiendo su cierre en un punto posterior del flujo de ejecución.

• Esto sólo funciona en evented servers

  Broadly speaking, there are two ways to handle concurrent requests to a server:

  1. Threaded servers use multiple concurrently-executing threads that each handle one client request,
  2. Evented servers run a single event loop that handles events for all connected clients
  3. Thin y Rainbows son ejemplos de servidores que pueden funcionar como evented servers. (Otros servidores quizá cerrarán el stream)

El siguiente ejemplo muestra como mantener una conexión persitente, abierta para enviar mensajes de broadcast a unos suscriptores:

[~/sinatra/sinatra-streaming/upandrunning-streaming(master)]$ cat a_simple_streaming_example.rb 
require 'sinatra'

before do
  content_type :txt
end

set server: 'thin', connections: []

get '/consume' do
  stream(:keep_open) do |out|
    # store connection for later on
    settings.connections << out
    logger.warn "connections.length = #{settings.connections.length}"

    # remove connection when closed properly
    out.callback do 
      logger.warn "connection closed. out = #{out}"
      settings.connections.delete(out) 
      logger.warn "connections.length = #{settings.connections.length}"
    end

    # remove connection when due to an error
    out.errback do
      logger.warn "we just lost  connection!. out = #{out}"
      settings.connections.delete(out)
      logger.warn "connections.length = #{settings.connections.length}"
    end
  end # stream
end

get '/produce/:message' do
  settings.connections.each do |out|
    out << "#{Time.now} -> #{params[:message]}" << "\n"
  end

  "Sent #{params[:message]} to all clients."
end

Para usar este ejemplo utilizaremos este Rakefile:

[~/sinatra/sinatra-streaming/upandrunning-streaming(master)]$ cat Rakefile 
task :default => :server

desc "run the server for the stream(:keep_open) example"
task :server do
  sh "ruby a_simple_streaming_example.rb"
end

desc "visit with browser 'localhost:4567/consume'"
task :consume do
  sh "open http://localhost:4567/consume"
end

desc "send messages to the consumer"
task :produce do
 (1..10).each do |i|
   sh "sleep 1; curl http://localhost:4567/produce/#{i}%0D"
 end
end

desc "start both consumer and producer"
task :all => [ :consume, :produce ]

1. Primero arrancamos el servidor. Obsérvese que el servidor usado es thin.

   [~/sinatra/sinatra-streaming/upandrunning-streaming(master)]$ rake server
   ruby a_simple_streaming_example.rb
   == Sinatra/1.4.4 has taken the stage on 4567 for development with backup from Thin
   Thin web server (v1.6.1 codename Death Proof)
   Maximum connections set to 1024
   Listening on localhost:4567, CTRL+C to stop
   

2. Después visitamos con un navegador la página localhost:4567/consume.

   [~/sinatra/sinatra-streaming/upandrunning-streaming(master)]$ rake consume
   open http://localhost:4567/consume
   

   Esto abre (en MacOS X) un navegador que queda a la espera del servidor de que la ruta de producción genere algún contenido
3. Si se desea abra alguna otra página de navegación privada (nueva ventana de incógnito en Chrome) en la misma URL localhost:4567/consume
4. Arranquemos el productor:

   [~/sinatra/sinatra-streaming/upandrunning-streaming(master)]$ rake produce
   sleep 1; curl http://localhost:4567/produce/1%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/2%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/3%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/4%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/5%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/6%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/7%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/8%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/9%0D
    to all clients.sleep 1; curl http://localhost:4567/produce/10%0D
    to all clients.
   

5. Esto hace que en (los) navegadores/clientes que estaban vistando localhost:4567/consume aparezca algo parecido a esto:

   2013-11-22 22:42:24 +0000 -> 1
   2013-11-22 22:42:25 +0000 -> 2
   2013-11-22 22:42:26 +0000 -> 3
   2013-11-22 22:42:27 +0000 -> 4
   2013-11-22 22:42:28 +0000 -> 5
   2013-11-22 22:42:29 +0000 -> 6
   2013-11-22 22:42:30 +0000 -> 7
   2013-11-22 22:42:31 +0000 -> 8
   2013-11-22 22:42:32 +0000 -> 9
   2013-11-22 22:42:33 +0000 -> 10
   

En la consola del servidor aparecerá algo parecido a esto:

[~/sinatra/sinatra-streaming/upandrunning-streaming(master)]$ rake
ruby a_simple_streaming_example.rb
== Sinatra/1.4.4 has taken the stage on 4567 for development with backup from Thin
Thin web server (v1.6.1 codename Death Proof)
Maximum connections set to 1024
Listening on localhost:4567, CTRL+C to stop
W, [2013-11-22T22:46:17.132773 #21927]  WARN -- : connections.length = 1
W, [2013-11-22T22:47:16.655453 #21927]  WARN -- : connection closed. out = #<Sinatra::Helpers::Stream:0x007ff10a9ed7b8>
W, [2013-11-22T22:47:16.655557 #21927]  WARN -- : connections.length = 0
W, [2013-11-22T22:47:16.655620 #21927]  WARN -- : we just lost  connection!. out = #<Sinatra::Helpers::Stream:0x007ff10a9ed7b8>
W, [2013-11-22T22:47:16.655677 #21927]  WARN -- : connections.length = 0
127.0.0.1 - - [22/Nov/2013 22:47:16] "GET /consume HTTP/1.1" 200 - 59.5292
127.0.0.1 - - [22/Nov/2013 22:50:32] "GET /produce/1%0D HTTP/1.1" 200 23 0.0009
127.0.0.1 - - [22/Nov/2013 22:50:33] "GET /produce/2%0D HTTP/1.1" 200 23 0.0008
127.0.0.1 - - [22/Nov/2013 22:50:34] "GET /produce/3%0D HTTP/1.1" 200 23 0.0009
127.0.0.1 - - [22/Nov/2013 22:50:35] "GET /produce/4%0D HTTP/1.1" 200 23 0.0008
127.0.0.1 - - [22/Nov/2013 22:50:36] "GET /produce/5%0D HTTP/1.1" 200 23 0.0009
127.0.0.1 - - [22/Nov/2013 22:50:37] "GET /produce/6%0D HTTP/1.1" 200 23 0.0009
127.0.0.1 - - [22/Nov/2013 22:50:38] "GET /produce/7%0D HTTP/1.1" 200 23 0.0009
127.0.0.1 - - [22/Nov/2013 22:50:39] "GET /produce/8%0D HTTP/1.1" 200 23 0.0007
127.0.0.1 - - [22/Nov/2013 22:50:40] "GET /produce/9%0D HTTP/1.1" 200 23 0.0006
127.0.0.1 - - [22/Nov/2013 22:50:41] "GET /produce/10%0D HTTP/1.1" 200 24 0.0009

El objeto out de la clase Sinatra::Helpers::Stream. dispone de los métodos #callback(&block) => Object y errback (parece que errback es un alias de callback).

# File 'lib/sinatra/base.rb'

def callback(&block)
  return yield if @closed
  @callbacks << block
end

1. Por debajo este ejemplo usa Event::Machine.
2. Event::Machine is a library for Ruby, C++, and Java programs.
3. It provides event-driven I/O using the Reactor pattern.
   1. The reactor design pattern is an event handling pattern for handling service requests delivered concurrently to a service handler by one or more inputs.
   2. The service handler then demultiplexes the incoming requests and dispatches them synchronously to the associated request handlers.
4. El módulo Module: EventMachine::Deferrable provee métodos (Object) callback(&block). y (Object) errback(&block)
   1. El método (Object) callback(&block) specifies a block to be executed if and when the Deferrable object receives a status of :succeeded.

          out.callback do 
            logger.warn "connection closed. out = #{out}"
            settings.connections.delete(out) 
            logger.warn "connections.length = #{settings.connections.length}"
          end
      

   2. El método (Object) errback(&block) specifies a block to be executed if and when the Deferrable object receives a status of :failed.

          out.errback do
            logger.warn "we just lost  connection!. out = #{out}"
            settings.connections.delete(out)
            logger.warn "connections.length = #{settings.connections.length}"
          end
      

   3. Event::Machine (EM) adds two different formalisms for lightweight concurrency to the Ruby programmer’s toolbox: spawned processes and deferrables.