<commit_before># encoding: utf-8

# Copyright 2014 <NAME>.
#
# This file is a modification of code from Logstash Forwarder.
# Copyright 2012-2013 <NAME> and contributors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

require 'thread'
require 'timeout'
require 'zlib'
require 'json'

module LogCourier
  class ShutdownSignal < StandardError; end
  class ProtocolError < StandardError; end

  # Implementation of the server
  class Server
    attr_reader :port

    def initialize(options = {})
      @options = {
        :logger => nil,
        :transport => 'tls'
      }.merge(options)

      @logger = @options[:logger]

      case @options[:transport]
      when 'tcp', 'tls'
        require 'log-courier/server_tcp'
        @server = ServerTcp.new(@options)
      when 'plainzmq', 'zmq'
        require 'log-courier/server_zmq'
        @server = ServerZmq.new(@options)
      else
        raise '[LogCourierServer] \'transport\' must be tcp, tls, plainzmq or zmq'
      end

      # Grab the port back
      @port = @server.port
    end

    def run(&block)
      # TODO: Make queue size configurable
      event_queue = SizedQueue.new 10
      spooler_thread = nil

      begin
        # Why a spooler thread? Well we don't know what &block is! We want connection threads to be non-blocking so they DON'T timeout
        # Non-blocking means we can keep clients informed of progress, and response in a timely fashion. We could create this with
        # a timeout wrapper around the &block call but we'd then be generating exceptions in someone else's code
        # So we allow the caller to block us - but only our spooler thread - our other threads are safe and we can use timeout
        spooler_thread = Thread.new do
          loop do
            events = event_queue.pop
            break if events.nil?
            events.each do |event|
              block.call event
            end<commit_msg>Add TODO note regarding logstash input plugin shutdown - relates to #30<commit_after># encoding: utf-8

# Copyright 2014 <NAME>.
#
# This file is a modification of code from Logstash Forwarder.
# Copyright 2012-2013 <NAME> and contributors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

require 'thread'
require 'timeout'
require 'zlib'
require 'json'

module LogCourier
  class ShutdownSignal < StandardError; end
  class ProtocolError < StandardError; end

  # Implementation of the server
  class Server
    attr_reader :port

    def initialize(options = {})
      @options = {
        :logger => nil,
        :transport => 'tls'
      }.merge(options)

      @logger = @options[:logger]

      case @options[:transport]
      when 'tcp', 'tls'
        require 'log-courier/server_tcp'
        @server = ServerTcp.new(@options)
      when 'plainzmq', 'zmq'
        require 'log-courier/server_zmq'
        @server = ServerZmq.new(@options)
      else
        raise '[LogCourierServer] \'transport\' must be tcp, tls, plainzmq or zmq'
      end

      # Grab the port back
      @port = @server.port
    end

    def run(&block)
      # TODO: Make queue size configurable
      event_queue = SizedQueue.new 10
      spooler_thread = nil

      begin
        # Why a spooler thread? Well we don't know what &block is! We want connection threads to be non-blocking so they DON'T timeout
        # Non-blocking means we can keep clients informed of progress, and response in a timely fashion. We could create this with
        # a timeout wrapper around the &block call but we'd then be generating exceptions in someone else's code
        # So we allow the caller to block us - but only our spooler thread - our other threads are safe and we can use timeout

        # TODO: Spooler thread becomes run() thread - then any exception falls through to the LogStash plugin
        # The only exception that raises in this small chunk of code should be a Logstash Shutdown exception
        # Therefore, when Logstash plugin receives the exception, it can call a teardown here which pulls
        # down the other threads with the correct exceptions
        # This resolves the problem where we want to keep all Logstash-ness outside of this library, and as a result
        # have problems with shutdown as we start receiving Logstash shutdown exceptions in here
        spooler_thread = Thread.new do
          loop do
            events = event_queue.pop
            break if events.nil?
            events.each do |event|
              block.call event
            end
{'fraction_non_alphanumeric': 0.06872600987022483, 'fraction_numerical': 0.007311277645768598, 'mean_word_length': 2.7790055248618786, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1042557', 'n_tokens_mistral': 1488, 'n_tokens_neox': 1415, 'n_words': 658}