Jonathan Gnagy's Blog

# ...
        # This methods tells the server to pull all logs from the LogQueue and 'process' them.
        # Essentially, this moves logs from the incoming LogQueue to a process ThreadPool for
        # controlled processing. This method prevents an insurge of messages from eating up all
        # the server's CPU by limiting the number of logs processed at a time. After the log is
        # parsed in the process pool, it is queued up for storage in the storage ThreadPool.
        def process_logs
                until @log_queue.empty?
                        @queue_flushed = false if @queue_flushed
                        log = pop_log
                        @process_pool.add_job {
                                puts "Processing Log: #{log.object_id.to_s(16)}"
                                parsed_log = @parser.parse log
                                @storage_pool.add_job {@storage.store_log parsed_log}
                        }
                end
                if !@queue_flushed
                        puts "Log Queue Flush Complete @ #{Time.now}" if !logs_queued?
                        @queue_flushed = true
                end
        end
        
        # Utilizes the LogMutex to synchronize initial storage of log packets in the LogQueue
        def enqueue_log(log_data)
                @mutex.synchronize { @log_queue.push log_data }
        end
        
        # Utilizes the LogMutex to synchronize removing log packets from the LogQueue
        def pop_log
                return @mutex.synchronize { @log_queue.pop }
        end
        
        # Utilizes the LogMutex to synchronize checking how many logs are currently queued in the
        # LogQueue.
        def logs_queued?
                return @mutex.synchronize { @log_queue.size > 0 }
        end
# ...