Как я могу нарезать и анализировать данные, полученные с помощью потокового API?

Question

Я пытаюсь подключиться к потоковому HTTP-интерфейсу Meetup и анализировать полученные события в разных записях. Я использую рубин над Синатрой. Я выбрал гем 'em-http-request' для обработки соединения и 'thin' в качестве сервера. В поисках информации о том, как обрабатывать API-потоки, все, что я нашел, было от 3 до 6 лет, и это лучший пример, который я нашел.

В этом примере автор использует регулярное выражение, чтобы найти конец каждого твита и разбить их на разные записи. В моем случае я не нашел способа разделить поток событий MeetUp.

Вот мой код:

get '/' do
STREAMING_URL = 'http://stream.meetup.com/2/open_events'
http = EM::HttpRequest.new(STREAMING_URL).get
buffer = ""
http.stream do |chunk|
    buffer += chunk
    while event = buffer.slice!(/{\"utc_offset\"+.../)
        eventRecord = event
        puts eventRecord
    end
end

Я открываю соединение, вызывающее //stream.meetup.com/2/open_events, и начинаю получать поток строк, случайно вырезанных в этом формате:

{"utc_offset":-14400000,"venue":{"country":"us","city":"Novi","address_1":"43155 Main St Suite 2300N","name":"Game of Clues Escape Room","lon":-83.470833,"state":"MI","lat":42.478107},"rsvp_limit":0,"venue_visibility":"public","visibility":"public","maybe_rsvp_count":0,"description":"<p>Search the Emerald City for clues to help you solve riddles and puzzles to escape the room before the 60 minute timer is up. Work together to complete missions that will bring your group closer together in order to get a clue card.<\/p> \n<p>25.00 per person. Book online at www.gameofclues.com<\/p>","mtime":1467030326494,"event_url":"http:\/\/www.meetup.com\/Escape-Room-Lovers\/events\/232071150\/","yes_rsvp_count":1,"duration":3600000,"payment_required":"0","name":"Game of Clues Escape Room Novi,Mi","id":"232071150","time":1467507600000,"group":{"join_mode":"open","country":"us","city":"Novi","name":"Escape Room Lovers","group_lon":-83.52,"id":20101745,"state":"MI","urlname":"Escape-Room-Lovers","category":{"name":"games","id":11,"shortname":"games"},"group_lat":42.47},"status":"upcoming"}{"utc_offset":14400000,"venue":{"country":"ae","city":"Dubai","address_1":"Jumeirah Lake Towers, outside Dubai Marina Metro","name":"Illuminations Well-Being Center, 409, Fortune Executive Towers, Cluster T, Plot T1, ","lon":55.311668,"lat":25.264444},"rsvp_limit":0,"venue_visibility":"public","visibility":"public","maybe_rsvp_count":0,"description":"<p>Facilitator: Dr. Beryl Bazley<\/p> \n<p>Investment: Free!<\/p> \n<p>For more information call

Я пытался использовать .slice! над содержимым буфера, используя в качестве параметра "{"utc_offset"", который является подстрокой, которая появляется в начале каждого события, но я не мог понять, как написать регулярное выражение, которое получает все, что заключено между каждым значением подстроки, чтобы получить как результат всего события.

Также я не уверен, что добавление чанков в буфер переменных и затем использование метода .slice! это лучший способ получить каждое событие.

Какой лучший способ решить эту ситуацию?

Как я могу нарезать и анализировать данные, полученные с помощью потокового API?

---

ЗДЕСЬ Я ДОБАВИЛ РЕАЛИЗАЦИЮ РЕШЕНИЯ, ПРЕДЛАГАЕМОГО @jordan В КОММЕНТАРИЯХ:

require 'yajl'
require 'uri' 
require 'yajl/http_stream'


@parser = Yajl::Parser.new(:symbolize_keys => true)

STREAMING_URL = 'http://stream.meetup.com/2/open_events'
Yajl::HttpStream.get(STREAMING_URL, :symbolize_keys => true) do |hash|
    puts hash.inspect
    hash.each {|key, value| puts "#{key} is #{value}" }
end