<commit_before>#
# stream_consumer_http.rb - This file contains the StreamConsumer_HTTP class.
#
# Copyright (C) 2011 MediaSift Ltd
#
# == Overview
#
# The StreamConsumer_HTTP class implements HTTP streaming.

$LOAD_PATH.unshift(File.dirname(__FILE__) + '/../')

require 'uri'
require 'socket'
require 'yajl'
require 'cgi'

module DataSift

	class StreamConsumer_HTTP < StreamConsumer

		# Constructor. Requires valid user and definition objects.
		def initialize(user, definition)
			super
		end

		def onStart(&block)
			begin
				reconnect() unless !@socket.nil? and !@socket.closed?

				parser = Yajl::Parser.new
				parser.on_parse_complete = block if block_given?
				if @response_head[:headers]["Transfer-Encoding"] == 'chunked'
					if block_given?
						chunkLeft = 0
						while !@socket.eof? && (line = @socket.gets)
							break if line.match /^0.*?\r\n/
							next if line == "\r\n"
							size = line.hex
							json = @socket.read(size)
							next if json.nil?
							chunkLeft = size-json.size
							if chunkLeft == 0
								if json.length > 100
									parser << json
								end
							else
								# received only part of the chunk, grab the rest
								parser << @socket.read(chunkLeft)
							end
						end
					else
						raise StreamError, 'Chunked responses detected, but no block given to handle the chunks.'
					end<commit_msg>Check if a close() has not been requested before looping<commit_after>#
# stream_consumer_http.rb - This file contains the StreamConsumer_HTTP class.
#
# Copyright (C) 2011 MediaSift Ltd
#
# == Overview
#
# The StreamConsumer_HTTP class implements HTTP streaming.

$LOAD_PATH.unshift(File.dirname(__FILE__) + '/../')

require 'uri'
require 'socket'
require 'yajl'
require 'cgi'

module DataSift

	class StreamConsumer_HTTP < StreamConsumer

		# Constructor. Requires valid user and definition objects.
		def initialize(user, definition)
			super
		end

		def onStart(&block)
			begin
				reconnect() unless !@socket.nil? and !@socket.closed?

				parser = Yajl::Parser.new
				parser.on_parse_complete = block if block_given?
				if @response_head[:headers]["Transfer-Encoding"] == 'chunked'
					if block_given?
						chunkLeft = 0
						while !@socket.eof? && (line = @socket.gets) && @state == StreamConsumer::STATE_RUNNING
							break if line.match /^0.*?\r\n/
							next if line == "\r\n"
							size = line.hex
							json = @socket.read(size)
							next if json.nil?
							chunkLeft = size-json.size
							if chunkLeft == 0
								if json.length > 100
									parser << json
								end
							else
								# received only part of the chunk, grab the rest
								parser << @socket.read(chunkLeft)
							end
						end
					else
						raise StreamError, 'Chunked responses detected, but no block given to handle the chunks.'
					end
{'fraction_non_alphanumeric': 0.10292022792022792, 'fraction_numerical': 0.007122507122507123, 'mean_word_length': 2.735372340425532, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '653846', 'n_tokens_mistral': 1157, 'n_tokens_neox': 837, 'n_words': 262}