From 0f32d26f4f666dcf382977df8974056f5b2e9ca8 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 00:21:49 +0000
Subject: [PATCH 01/46] Simplify the version generation and fix dirty builds
 appearing after first make (helps #62)

---
 build/fix_version | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/build/fix_version b/build/fix_version
index 36ae1f76..666b0947 100755
--- a/build/fix_version
+++ b/build/fix_version
@@ -8,9 +8,12 @@ fi
 if [ -n "$1" ]; then
 	VERSION="$1"
 else
-	VERSION="$(git describe)"
-	if [ $(git status -s 2>/dev/null | wc -l) -ne 0 ]; then
-		VERSION="${VERSION}-dirty"
+	# Describe version from Git, and ensure the only "-xxx" is the git revision
+	# This ensures that gem builds only add one ".pre" tag automatically
+	VERSION="$(git describe | sed 's/-\([0-9][0-9]*\)-\([0-9a-z][0-9a-z]*\)$/-\1\2/g')"
+	# Append "D" to show a dirty build, but ignore changes to versions in gemspecs etc.
+	if [ $(git status -s 2>/dev/null | grep -vE '\s(log-courier.gemspec|src/lc-lib/core/version.go)$' | wc -l) -ne 0 ]; then
+		VERSION="${VERSION}D"
 	fi
 fi
 

From dc6113035bd1d0752ad2dcdcc75a2a715bc8149f Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 00:25:13 +0000
Subject: [PATCH 02/46] Draft implementation of logstash gem generation and
 publication (Fixes #60) Improved version patching

---
 .gitignore                          |  2 +-
 Gemfile                             | 14 ++++++++++++--
 Makefile                            | 12 ++++++++----
 Rakefile                            | 13 +++++++++++++
 build/fix_version                   | 15 +++++++++++++--
 log-courier.gemspec                 |  4 ++--
 log-courier.gemspec.template        | 23 -----------------------
 logstash-input-log-courier.gemspec  | 20 ++++++++++++++++++++
 logstash-output-log-courier.gemspec | 20 ++++++++++++++++++++
 9 files changed, 89 insertions(+), 34 deletions(-)
 create mode 100644 Rakefile
 delete mode 100644 log-courier.gemspec.template
 create mode 100644 logstash-input-log-courier.gemspec
 create mode 100644 logstash-output-log-courier.gemspec

diff --git a/.gitignore b/.gitignore
index 7e1ab385..56a6fe27 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,5 @@
 Gemfile.lock
 bin
-log-courier-*.gem
 node_modules
 pkg
 spec/tmp
@@ -10,3 +9,4 @@ vendor
 .bundle
 .log-courier
 .vagrant
+*.gem
diff --git a/Gemfile b/Gemfile
index 164ccb2a..4b5a27a8 100644
--- a/Gemfile
+++ b/Gemfile
@@ -1,9 +1,19 @@
 source 'https://rubygems.org'
+
+# Log Courier gem dependencies
 gem 'ffi-rzmq'
 gem 'multi_json'
-gem 'rspec'
 
+# Log Courier gem JSON parsers
 gem 'oj',        :platforms => :mri
+gem 'jrjackson', :platforms => :jruby
+
+# Profiler for MRI
 gem 'ruby-prof', :platforms => :mri
 
-gem 'jrjackson', :platforms => :jruby
+# Tests
+gem 'rspec'
+
+# Publishing to ruby gems
+gem 'rake'
+gem 'gem_publisher'
diff --git a/Makefile b/Makefile
index bd30faef..fff92e10 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: prepare all log-courier gem test doc profile benchmark jrprofile jrbenchmark clean
+.PHONY: prepare fix_version all log-courier gem test doc profile benchmark jrprofile jrbenchmark clean
 
 MAKEFILE := $(word $(words $(MAKEFILE_LIST)),$(MAKEFILE_LIST))
 GOPATH := $(patsubst %/,%,$(dir $(abspath $(MAKEFILE))))
@@ -36,8 +36,10 @@ all: log-courier
 
 log-courier: $(BINS)
 
-gem:
+gem: | fix_version
 	gem build log-courier.gemspec
+	gem build logstash-input-log-courier.gemspec
+	gem build logstash-output-log-courier.gemspec
 
 test: all vendor/bundle/.GemfileModT
 	go get -d -tags "$(TAGS)" $(GOTESTS)
@@ -84,11 +86,13 @@ ifneq ($(implyclean),yes)
 	rm -f log-courier-*.gem
 endif
 
-prepare:
+fix_version:
+	build/fix_version
+
+prepare: | fix_version
 	@go version >/dev/null || (echo "Go not found. You need to install Go version 1.2 or 1.3: http://golang.org/doc/install"; false)
 	@go version | grep -q 'go version go1.[23]' || (echo "Go version 1.2 or 1.3 required, you have a version of Go that is not supported."; false)
 	@echo "GOPATH: $${GOPATH}"
-	build/fix_version
 
 bin/%: FORCE | prepare
 	go get -d -tags "$(TAGS)" $*
diff --git a/Rakefile b/Rakefile
new file mode 100644
index 00000000..d1894959
--- /dev/null
+++ b/Rakefile
@@ -0,0 +1,13 @@
+task :default do
+  system('rake -T')
+end
+
+# Publish gem to rubygems
+# https://github.com/logstash-plugins/logstash-input-tcp/blob/master/rakelib/publish.rake
+desc 'Publish gem to RubyGems.org'
+task :publish_gem do
+  require 'gem_publisher'
+  gem_file = Dir.glob(File.expand_path('*.gemspec', File.dirname(__FILE__))).first
+  gem = GemPublisher.publish_if_updated(gem_file, :rubygems)
+  puts "Published #{gem}" if gem
+end
diff --git a/build/fix_version b/build/fix_version
index 666b0947..15c1be9e 100755
--- a/build/fix_version
+++ b/build/fix_version
@@ -17,7 +17,18 @@ else
 	fi
 fi
 
-sed "s/<VERSION>/${VERSION}/g" src/lc-lib/core/version.go.template > src/lc-lib/core/version.go
-sed "s/<VERSION>/${VERSION#v}/g" log-courier.gemspec.template > log-courier.gemspec
+# Patch version.go
+sed "s/\\(const  *Log_Courier_Version  *string  *=  *\"\\)[^\"]*\\(\"\\)/\\1${VERSION}\\2/g" src/lc-lib/core/version.go > src/lc-lib/core/version.go.tmp
+\mv -f src/lc-lib/core/version.go.tmp src/lc-lib/core/version.go
+
+# Patch the gemspecs
+for GEM in log-courier logstash-input-log-courier logstash-output-log-courier; do
+	sed "s/\\(gem.version *= *'\\)[^']*\\('\\)/\\1${VERSION#v}\\2/g" ${GEM}.gemspec > ${GEM}.gemspec.tmp
+	\mv -f ${GEM}.gemspec.tmp ${GEM}.gemspec
+	[ ${GEM#logstash-} != $GEM ] && {
+		sed "s/\\(gem.add_runtime_dependency  *'log-courier' *,  *'= *\\)[^']*\\('\\)/\\1${VERSION#v}\\2/g" ${GEM}.gemspec > ${GEM}.gemspec.tmp
+		\mv -f ${GEM}.gemspec.tmp ${GEM}.gemspec
+	}
+done
 
 echo "Setting Log Courier Version ${VERSION}"
diff --git a/log-courier.gemspec b/log-courier.gemspec
index 8beb7d26..2218a076 100644
--- a/log-courier.gemspec
+++ b/log-courier.gemspec
@@ -18,6 +18,6 @@ Gem::Specification.new do |gem|
     lib/log-courier/server_zmq.rb
   )
 
-  gem.add_runtime_dependency 'ffi-rzmq', '>= 2.0'
-  gem.add_runtime_dependency 'multi_json'
+  gem.add_runtime_dependency 'ffi-rzmq', '~> 2.0'
+  gem.add_runtime_dependency 'multi_json', '~> 1.0'
 end
diff --git a/log-courier.gemspec.template b/log-courier.gemspec.template
deleted file mode 100644
index 543b4637..00000000
--- a/log-courier.gemspec.template
+++ /dev/null
@@ -1,23 +0,0 @@
-Gem::Specification.new do |gem|
-  gem.name              = 'log-courier'
-  gem.version           = '<VERSION>'
-  gem.description       = 'Log Courier library'
-  gem.summary           = 'Receive events from Log Courier and transmit between LogStash instances'
-  gem.homepage          = 'https://github.com/driskell/log-courier'
-  gem.authors           = ['Jason Woods']
-  gem.email             = ['devel@jasonwoods.me.uk']
-  gem.licenses          = ['Apache']
-  gem.rubyforge_project = 'nowarning'
-  gem.require_paths     = ['lib']
-  gem.files             = %w(
-    lib/log-courier/client.rb
-    lib/log-courier/client_tls.rb
-    lib/log-courier/event_queue.rb
-    lib/log-courier/server.rb
-    lib/log-courier/server_tcp.rb
-    lib/log-courier/server_zmq.rb
-  )
-
-  gem.add_runtime_dependency 'ffi-rzmq', '>= 2.0'
-  gem.add_runtime_dependency 'multi_json'
-end
diff --git a/logstash-input-log-courier.gemspec b/logstash-input-log-courier.gemspec
new file mode 100644
index 00000000..185a4426
--- /dev/null
+++ b/logstash-input-log-courier.gemspec
@@ -0,0 +1,20 @@
+Gem::Specification.new do |gem|
+  gem.name              = 'logstash-input-log-courier'
+  gem.version           = '1.0'
+  gem.description       = 'Log Courier Input Logstash Plugin'
+  gem.summary           = 'Receive events from Log Courier and Logstash using the Log Courier protocol'
+  gem.homepage          = 'https://github.com/driskell/log-courier'
+  gem.authors           = ['Jason Woods']
+  gem.email             = ['devel@jasonwoods.me.uk']
+  gem.licenses          = ['Apache']
+  gem.rubyforge_project = 'nowarning'
+  gem.require_paths     = ['lib']
+  gem.files             = %w(
+    lib/logstash/inputs/courier.rb
+  )
+
+  gem.metadata = { 'logstash_plugin' => 'true', 'group' => 'input' }
+
+  gem.add_runtime_dependency 'logstash', '~> 1.4'
+  gem.add_runtime_dependency 'log-courier', '= 1.0'
+end
diff --git a/logstash-output-log-courier.gemspec b/logstash-output-log-courier.gemspec
new file mode 100644
index 00000000..b9ec8075
--- /dev/null
+++ b/logstash-output-log-courier.gemspec
@@ -0,0 +1,20 @@
+Gem::Specification.new do |gem|
+  gem.name              = 'logstash-output-log-courier'
+  gem.version           = '1.0'
+  gem.description       = 'Log Courier Output Logstash Plugin'
+  gem.summary           = 'Transmit events from one Logstash instance to another using the Log Courier protocol'
+  gem.homepage          = 'https://github.com/driskell/log-courier'
+  gem.authors           = ['Jason Woods']
+  gem.email             = ['devel@jasonwoods.me.uk']
+  gem.licenses          = ['Apache']
+  gem.rubyforge_project = 'nowarning'
+  gem.require_paths     = ['lib']
+  gem.files             = %w(
+    lib/logstash/outputs/courier.rb
+  )
+
+  gem.metadata = { 'logstash_plugin' => 'true', 'group' => 'input' }
+
+  gem.add_runtime_dependency 'logstash', '~> 1.4'
+  gem.add_runtime_dependency 'log-courier', '= 1.0'
+end

From 8d7adeb414f451957eefd4d9ce22d72b6d8ee6a3 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 00:50:17 +0000
Subject: [PATCH 03/46] Print ZMQ version information in logs

---
 lib/log-courier/server_zmq.rb | 8 ++++++++
 src/lc-lib/transports/zmq.go  | 3 +++
 2 files changed, 11 insertions(+)

diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index 04192cd7..e6a26c7d 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -14,7 +14,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+require 'ffi-rzmq-core'
+require 'ffi-rzmq-core/version'
 require 'ffi-rzmq'
+require 'ffi-rzmq/version'
 
 module LogCourier
   # ZMQ transport implementation for the server
@@ -73,6 +76,11 @@ def initialize(options = {})
         raise "[LogCourierServer] Failed to initialise: #{e}"
       end
 
+      libversion = LibZMQ.version
+      @logger.info "[LogCourierServer] libzmq version #{libversion[:major]}.#{libversion[:minor]}.#{libversion[:patch]}"
+      @logger.info "[LogCourierServer] ffi-rzmq-core version #{LibZMQ::VERSION}"
+      @logger.info "[LogCourierServer] ffi-rzmq version #{ZMQ.version}"
+
       # TODO: Implement workers option by receiving on a ROUTER and proxying to a DEALER, with workers connecting to the DEALER
 
       reset_timeout
diff --git a/src/lc-lib/transports/zmq.go b/src/lc-lib/transports/zmq.go
index 27c53756..086899dc 100644
--- a/src/lc-lib/transports/zmq.go
+++ b/src/lc-lib/transports/zmq.go
@@ -289,6 +289,9 @@ func (t *TransportZmq) Init() (err error) {
     return errors.New("Failed to register any of the specified endpoints.")
   }
 
+  major, minor, patch := zmq.Version()
+  log.Info("libzmq version %d.%d.%d", major, minor, patch)
+
   // Signal channels
   t.bridge_chan = make(chan []byte, 1)
   t.send_chan = make(chan *ZMQMessage, 2)

From c35b9b16c699a41734d30ff09be1f403899ebb74 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 10:25:10 +0000
Subject: [PATCH 04/46] Make develop version compatible with older rubygems
 versions, specifically v1.8

---
 build/fix_version | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/build/fix_version b/build/fix_version
index 15c1be9e..abccad94 100755
--- a/build/fix_version
+++ b/build/fix_version
@@ -10,7 +10,7 @@ if [ -n "$1" ]; then
 else
 	# Describe version from Git, and ensure the only "-xxx" is the git revision
 	# This ensures that gem builds only add one ".pre" tag automatically
-	VERSION="$(git describe | sed 's/-\([0-9][0-9]*\)-\([0-9a-z][0-9a-z]*\)$/-\1\2/g')"
+	VERSION="$(git describe | sed 's/-\([0-9][0-9]*\)-\([0-9a-z][0-9a-z]*\)$/.\1.\2/g')"
 	# Append "D" to show a dirty build, but ignore changes to versions in gemspecs etc.
 	if [ $(git status -s 2>/dev/null | grep -vE '\s(log-courier.gemspec|src/lc-lib/core/version.go)$' | wc -l) -ne 0 ]; then
 		VERSION="${VERSION}D"

From 5f410cf6b4a8d00bf5219094f645496b891446be Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 10:28:01 +0000
Subject: [PATCH 05/46] Update change log

---
 docs/ChangeLog.md | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/docs/ChangeLog.md b/docs/ChangeLog.md
index 4be8d638..02cdde2b 100644
--- a/docs/ChangeLog.md
+++ b/docs/ChangeLog.md
@@ -4,6 +4,7 @@
 <!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
 **Table of Contents**  *generated with [DocToc](http://doctoc.herokuapp.com/)*
 
+- [Latest Development](#latest-development)
 - [1.0](#10)
 - [0.15](#015)
 - [0.14](#014)
@@ -16,6 +17,16 @@
 
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
 
+## Latest Development
+
+*TBC*
+
+* Implement gems for the new Logstash plugin system (#60)
+* Fix gem build failing on develop branch with old rubygems versions due to a
+malformed version string (#62)
+* Print informational messages containing ZMQ library version information during
+gem and log-courier startup to aid in diagnostics
+
 ## 1.0
 
 *23rd October 2014*

From 4fa452860f19ab8faba3324f203d9dc40fd3e5f9 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 10:37:40 +0000
Subject: [PATCH 06/46] Fix zmq_3_x build (towards #63)

---
 src/lc-lib/transports/zmq.go             | 16 ++--------------
 src/lc-lib/transports/zmq_curve.go       | 20 ++++++++++++++++++++
 src/lc-lib/transports/zmq_nocurve.go     |  4 ++++
 src/lc-lib/transports/zmq_z85validate.go |  2 +-
 4 files changed, 27 insertions(+), 15 deletions(-)

diff --git a/src/lc-lib/transports/zmq.go b/src/lc-lib/transports/zmq.go
index 086899dc..55f56cc7 100644
--- a/src/lc-lib/transports/zmq.go
+++ b/src/lc-lib/transports/zmq.go
@@ -117,20 +117,8 @@ func NewZmqTransportFactory(config *core.Config, config_path string, unused map[
       return nil, err
     }
 
-    if len(ret.CurveServerkey) == 0 {
-      return nil, fmt.Errorf("Option %scurve server key is required", config_path)
-    } else if len(ret.CurveServerkey) != 40 || !z85Validate(ret.CurveServerkey) {
-      return nil, fmt.Errorf("Option %scurve server key must be a valid 40 character Z85 encoded string", config_path)
-    }
-    if len(ret.CurvePublickey) == 0 {
-      return nil, fmt.Errorf("Option %scurve public key is required", config_path)
-    } else if len(ret.CurvePublickey) != 40 || !z85Validate(ret.CurvePublickey) {
-      return nil, fmt.Errorf("Option %scurve public key must be a valid 40 character Z85 encoded string", config_path)
-    }
-    if len(ret.CurveSecretkey) == 0 {
-      return nil, fmt.Errorf("Option %scurve secret key is required", config_path)
-    } else if len(ret.CurveSecretkey) != 40 || !z85Validate(ret.CurveSecretkey) {
-      return nil, fmt.Errorf("Option %scurve secret key must be a valid 40 character Z85 encoded string", config_path)
+    if err := ret.processConfig(config_path); err != nil {
+      return nil, err
     }
   } else {
     if err := config.ReportUnusedConfig(config_path, unused); err != nil {
diff --git a/src/lc-lib/transports/zmq_curve.go b/src/lc-lib/transports/zmq_curve.go
index cfe08f53..a7c03315 100644
--- a/src/lc-lib/transports/zmq_curve.go
+++ b/src/lc-lib/transports/zmq_curve.go
@@ -23,6 +23,26 @@ import (
   "lc-lib/core"
 )
 
+func (f *TransportZmqFactory) processConfig(config_path string) (err error) {
+  if len(f.CurveServerkey) == 0 {
+    return fmt.Errorf("Option %scurve server key is required", config_path)
+  } else if len(f.CurveServerkey) != 40 || !z85Validate(f.CurveServerkey) {
+    return fmt.Errorf("Option %scurve server key must be a valid 40 character Z85 encoded string", config_path)
+  }
+  if len(f.CurvePublickey) == 0 {
+    return fmt.Errorf("Option %scurve public key is required", config_path)
+  } else if len(f.CurvePublickey) != 40 || !z85Validate(f.CurvePublickey) {
+    return fmt.Errorf("Option %scurve public key must be a valid 40 character Z85 encoded string", config_path)
+  }
+  if len(f.CurveSecretkey) == 0 {
+    return fmt.Errorf("Option %scurve secret key is required", config_path)
+  } else if len(f.CurveSecretkey) != 40 || !z85Validate(f.CurveSecretkey) {
+    return fmt.Errorf("Option %scurve secret key must be a valid 40 character Z85 encoded string", config_path)
+  }
+
+  return nil
+}
+
 func (t *TransportZmq) configureSocket() (err error) {
   if t.config.transport == "zmq" {
     // Configure CurveMQ security
diff --git a/src/lc-lib/transports/zmq_nocurve.go b/src/lc-lib/transports/zmq_nocurve.go
index e5f580b5..ce39bb07 100644
--- a/src/lc-lib/transports/zmq_nocurve.go
+++ b/src/lc-lib/transports/zmq_nocurve.go
@@ -18,6 +18,10 @@
 
 package transports
 
+func (f *TransportZmqFactory) processConfig(config_path string) (err error) {
+  return nil
+}
+
 func (t *TransportZmq) configureSocket() error {
   return nil
 }
diff --git a/src/lc-lib/transports/zmq_z85validate.go b/src/lc-lib/transports/zmq_z85validate.go
index 3d5d826d..674b7618 100644
--- a/src/lc-lib/transports/zmq_z85validate.go
+++ b/src/lc-lib/transports/zmq_z85validate.go
@@ -1,4 +1,4 @@
-// +build zmq
+// +build zmq_4_x
 
 /*
  * Copyright 2014 Jason Woods.

From ab858fb7982481cd7a4003dafa65ea558a39b5c3 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 10:46:43 +0000
Subject: [PATCH 07/46] Graceful exit if using zmq transport with libzmq <= 4

---
 docs/ChangeLog.md             | 2 ++
 lib/log-courier/server_zmq.rb | 8 ++++++--
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/docs/ChangeLog.md b/docs/ChangeLog.md
index 02cdde2b..66f9b65b 100644
--- a/docs/ChangeLog.md
+++ b/docs/ChangeLog.md
@@ -26,6 +26,8 @@
 malformed version string (#62)
 * Print informational messages containing ZMQ library version information during
 gem and log-courier startup to aid in diagnostics
+* Raise a friendly error when trying to use the zmq transport in the Log Courier
+gem with incompatible versions of libzmq
 
 ## 1.0
 
diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index e6a26c7d..2d09f8f5 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -38,7 +38,12 @@ def initialize(options = {})
 
       @logger = @options[:logger]
 
+      libversion = LibZMQ.version
+      libversion = "#{libversion[:major]}.#{libversion[:minor]}.#{libversion[:patch]}"
+
       if @options[:transport] == 'zmq'
+        raise "[LogCourierServer] Transport 'zmq' requires libzmq version >= 4 (the current version is #{libversion})" unless LibZMQ.version4?
+
         raise '[LogCourierServer] \'curve_secret_key\' is required' if @options[:curve_secret_key].nil?
 
         raise '[LogCourierServer] \'curve_secret_key\' must be a valid 40 character Z85 encoded string' if @options[:curve_secret_key].length != 40 || !z85validate(@options[:curve_secret_key])
@@ -76,8 +81,7 @@ def initialize(options = {})
         raise "[LogCourierServer] Failed to initialise: #{e}"
       end
 
-      libversion = LibZMQ.version
-      @logger.info "[LogCourierServer] libzmq version #{libversion[:major]}.#{libversion[:minor]}.#{libversion[:patch]}"
+      @logger.info "[LogCourierServer] libzmq version #{libversion}"
       @logger.info "[LogCourierServer] ffi-rzmq-core version #{LibZMQ::VERSION}"
       @logger.info "[LogCourierServer] ffi-rzmq version #{ZMQ.version}"
 

From 3f2fee612db1b90e27eecdc80bc29f9a293cabbe Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 11:03:52 +0000
Subject: [PATCH 08/46] Tidy makefile and add bundle rake call for publishing
 gems

---
 Makefile | 21 +++++++++++----------
 1 file changed, 11 insertions(+), 10 deletions(-)

diff --git a/Makefile b/Makefile
index fff92e10..fee4829b 100644
--- a/Makefile
+++ b/Makefile
@@ -32,16 +32,19 @@ SAVETAGS := $(shell echo "$(TAGS)" >.Makefile.tags)
 endif
 endif
 
-all: log-courier
+all: | log-courier
 
-log-courier: $(BINS)
+log-courier: | $(BINS)
 
 gem: | fix_version
 	gem build log-courier.gemspec
 	gem build logstash-input-log-courier.gemspec
 	gem build logstash-output-log-courier.gemspec
 
-test: all vendor/bundle/.GemfileModT
+publish_gem: | fix_version vendor/bundle/.GemfileModT
+	bundle exec rake publish_gem
+
+test: | all vendor/bundle/.GemfileModT
 	go get -d -tags "$(TAGS)" $(GOTESTS)
 	go test -tags "$(TAGS)" $(GOTESTS)
 	bundle exec rspec $(TESTS)
@@ -58,20 +61,20 @@ doc:
 	@node_modules/.bin/doctoc README.md
 	@for F in docs/*.md docs/codecs/*.md; do node_modules/.bin/doctoc $$F; done
 
-profile: all vendor/bundle/.GemfileModT
+profile: | all vendor/bundle/.GemfileModT
 	bundle exec rspec spec/profile_spec.rb
 
-benchmark: all vendor/bundle/.GemfileModT
+benchmark: | all vendor/bundle/.GemfileModT
 	bundle exec rspec spec/benchmark_spec.rb
 
 vendor/bundle/.GemfileModT: Gemfile
 	bundle install --path vendor/bundle
 	@touch $@
 
-jrprofile: all vendor/bundle/.GemfileModT
+jrprofile: | all vendor/bundle/.GemfileModT
 	jruby --profile -G vendor/bundle/jruby/1.9/bin/rspec spec/benchmark_spec.rb
 
-jrbenchmark: all vendor/bundle/.GemfileJRubyModT
+jrbenchmark: | all vendor/bundle/.GemfileJRubyModT
 	jruby -G vendor/bundle/jruby/1.9/bin/rspec spec/benchmark_spec.rb
 
 vendor/bundle/.GemfileJRubyModT: Gemfile
@@ -94,8 +97,6 @@ prepare: | fix_version
 	@go version | grep -q 'go version go1.[23]' || (echo "Go version 1.2 or 1.3 required, you have a version of Go that is not supported."; false)
 	@echo "GOPATH: $${GOPATH}"
 
-bin/%: FORCE | prepare
+bin/%: prepare
 	go get -d -tags "$(TAGS)" $*
 	go install -tags "$(TAGS)" $*
-
-FORCE:

From 5f03d0928d95bc71dc870d85421337b91b102c9d Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 11:16:58 +0000
Subject: [PATCH 09/46] Wrap zeromq load error in gem with our own exception so
 the user knows it's us that failed

---
 lib/log-courier/server_zmq.rb | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index 2d09f8f5..2d869e20 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -14,10 +14,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-require 'ffi-rzmq-core'
-require 'ffi-rzmq-core/version'
-require 'ffi-rzmq'
-require 'ffi-rzmq/version'
+begin
+  require 'ffi-rzmq-core'
+  require 'ffi-rzmq-core/version'
+  require 'ffi-rzmq'
+  require 'ffi-rzmq/version'
+rescue LoadError => e
+  raise "[LogCourierServer] Could not initialise: #{e}"
+end
 
 module LogCourier
   # ZMQ transport implementation for the server

From 7a62cd547b6bfc9e7b40f4f48cd708116be62590 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 12:23:58 +0000
Subject: [PATCH 10/46] Improve documentation - make ZMQ less prominent due to
 the additional dependencies, and document the Logstash 1.5.x plugin manager
 approach

---
 README.md                   | 51 +++++++++++++++++++++----------
 docs/LogstashIntegration.md | 60 +++++++++++++++++++++++--------------
 2 files changed, 72 insertions(+), 39 deletions(-)

diff --git a/README.md b/README.md
index d66bf0ca..f2b99429 100644
--- a/README.md
+++ b/README.md
@@ -12,9 +12,10 @@ with many fixes and behavioural improvements.
 
 - [Features](#features)
 - [Installation](#installation)
-  - [Build Requirements](#build-requirements)
+  - [Requirements](#requirements)
   - [Building](#building)
   - [Logstash Integration](#logstash-integration)
+  - [Building with ZMQ support](#building-with-zmq-support)
   - [Generating Certificates and Keys](#generating-certificates-and-keys)
 - [Documentation](#documentation)
 
@@ -48,14 +49,12 @@ two Logstash instances.
 
 ## Installation
 
-### Build Requirements
+### Requirements
 
 1. \*nix, OS X or Windows
 1. The [golang](http://golang.org/doc/install) compiler tools (1.2 or 1.3)
 1. [git](http://git-scm.com)
 1. GNU make
-1. (Optional) [ZeroMQ](http://zeromq.org/intro:get-the-software) (>=3.2 for
-plaintext ZMQ, >=4.0 for secure CurveZMQ)
 
 *\*nix: Most requirements can usually be installed by your favourite package
 manager. The optional ZeroMQ >=3.2 is usually also available via the package
@@ -79,15 +78,6 @@ follows.
 
 The log-courier program can then be found in the 'bin' folder.
 
-To build with the optional ZMQ support use the following.
-
-	git clone https://github.com/driskell/log-courier
-	cd log-courier
-	make with=zmq3
-
-For CurveZMQ support (ZMQ with public key encryption) replace `zmq3` with
-`zmq4`.
-
 *If you receive errors whilst running `make` try `gmake` instead.*
 
 ### Logstash Integration
@@ -96,8 +86,36 @@ Log Courier does not utilise the lumberjack Logstash plugin and instead uses its
 own custom plugin. This allows significant enhancements to the integration far
 beyond the lumberjack protocol allows.
 
-Details instructions on the plugin and how to install it into Logstash can be
-found on the [Logstash Integration](docs/LogstashIntegration.md) page.
+Install using the Logstash 1.5+ Plugin manager.
+
+	cd /path/to/logstash
+	bin/logstash plugin install logstash-input-log-courier
+
+Detailed instructions, including integration with Logstash 1.4.x, can be found
+on the [Logstash Integration](docs/LogstashIntegration.md) page.
+
+### Building with ZMQ support
+
+To use the 'zmq' and 'plainzmq' transports, you will need to install
+[ZeroMQ](http://zeromq.org/intro:get-the-software) (>=3.2 for cleartext
+plainzmq, >=4.0 for encrypted zmq).
+
+*\*nix: ZeroMQ >=3.2 is usually available via the package manager. ZeroMQ >=4.0
+may need to be built and installed manually.*
+
+*OS X: ZeroMQ can be installed via [Homebrew](http://brew.sh).*
+
+*Windows: ZeroMQ will need to be built and installed manually.*
+
+Once the required version of ZeroMQ is installed, run the corresponding `make`
+command to build Log Courier with the ZMQ transports.
+
+	# ZeroMQ >=3.2 - cleartext 'plainzmq' transport
+	make with=zmq3
+	# ZeroMQ >=4.0 - both cleartext 'plainzmq' and encrypted 'zmq' transport
+	make with=zmq4
+
+*If you receive errors whilst running `make` try `gmake` instead.*
 
 ### Generating Certificates and Keys
 
@@ -107,7 +125,8 @@ TLS shipping transport.
 
 Likewise, running `make curvekey` will automatically build and run the
 `lc-curvekey` utility that can quickly and easily generate CurveZMQ key pairs
-for the CurveZMQ shipping transport.
+for the CurveZMQ shipping transport. This tool is only available when Log
+Courier is built with ZeroMQ >=4.0.
 
 Both tools also generate the required configuration file snippets.
 
diff --git a/docs/LogstashIntegration.md b/docs/LogstashIntegration.md
index 75ec29c2..65169898 100644
--- a/docs/LogstashIntegration.md
+++ b/docs/LogstashIntegration.md
@@ -8,52 +8,66 @@ Log Courier is built to work seamlessly with [Logstash](http://logstash.net)
 **Table of Contents**  *generated with [DocToc](http://doctoc.herokuapp.com/)*
 
 - [Installation](#installation)
-- [Remote Installation](#remote-installation)
+  - [Logstash 1.5+ Plugin Manager](#logstash-15-plugin-manager)
+  - [Manual installation](#manual-installation)
+  - [Local-only Installation](#local-only-installation)
 - [Configuration](#configuration)
 
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
 
 ## Installation
 
-To enable communication with Logstash the ruby gem needs to be installed into
-the Logstash installation, and then the input and output plugins.
+### Logstash 1.5+ Plugin Manager
 
-The following instructions assume you are using the tar.gz or packaged Logstash
-installations and that Logstash is installed to /opt/logstash. You should change
-this path if yours is different.
+Logstash 1.5 introduces a new plugin manager that makes installing additional
+plugins extremely easy.
 
-First build the gem. This will generate a file called log-courier-X.X.gem.
+Simply run the following commands to install the latest stable version of the
+Log Courier plugins. If you are only receiving events, you only need to install
+the input plugin.
+
+		cd /path/to/logstash
+		bin/logstash plugin install logstash-input-log-courier
+		bin/logstash plugin install logstash-output-log-courier
+
+Once the installation is complete, you can start using the plugins!
+
+### Manual installation
+
+For Logstash 1.4.x the plugins and dependencies need to be installed manually.
+
+First build the Log Courier gem the plugins require. The file you will need will
+be called log-courier-X.X.gem, where X.X is the version of Log Courier you have.
 
 		git clone https://github.com/driskell/log-courier
 		cd log-courier
 		make gem
 
-Then switch to the Logstash installation directory and install it. Note that
-because this is JRuby it may take a minute to finish the install. The
-ffi-rzmq-core and ffi-rzmq gems bundled with Logstash will be upgraded during
-the installation, which will require an internet connection.
+Switch to the Logstash installation directory and install it. Note that because
+this is JRuby it may take a minute to finish the install. The ffi-rzmq-core and
+ffi-rzmq gems bundled with Logstash will be upgraded during the installation,
+which will require an internet connection.
 
-		cd /opt/logstash
+		cd /path/to/logstash
 		export GEM_HOME=vendor/bundle/jruby/1.9
-		java -jar vendor/jar/jruby-complete-1.7.11.jar -S gem install <path-to-gem>
+		java -jar vendor/jar/jruby-complete-1.7.11.jar -S gem install /path/to/log-courier-X.X.gem
 
-Now install the Logstash plugins.
+The remaining step is to manually install the Logstash plugins.
 
-		cd <log-courier-source>
-		cp -rvf lib/logstash /opt/logstash/lib
+		cd /path/to/log-courier
+		cp -rvf lib/logstash /path/to/logstash/lib
 
-## Remote Installation
+### Local-only Installation
 
-If you need to install the gem on a server without an internet connection, you
-must download the latest ffi-rzmq-core and ffi-zmq gems from the rubygems site
-and transfer them across.
+If you need to install the gem and plugins on a server without an internet
+connection, you can download the latest ffi-rzmq-core and ffi-zmq gems from the
+rubygems site, transfer them across, and install them yourself. Once they are
+installed, follow the instructions for Manual Installation and the process can
+be completed without an internet connection.
 
 * https://rubygems.org/gems/ffi-rzmq-core
 * https://rubygems.org/gems/ffi-rzmq
 
-These gems should be installed before the log-courer gem. As a result, the
-log-courier gem installation will not need them to be downloaded.
-
 ## Configuration
 
 The 'courier' input and output plugins will now be available. An example

From 64aab9ff7004ceea3985464854b48d2bca44e97a Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 12:55:42 +0000
Subject: [PATCH 11/46] Remove rvm get stable which may not be required and is
 breaking travis currently

---
 .travis.yml | 1 -
 1 file changed, 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index 3a841ab2..479e4fba 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -8,7 +8,6 @@ go:
 # https://github.com/travis-ci/travis-ci/issues/2220
 # Tests require ruby
 install:
-  - rvm get stable
   - rvm use 2.0 --install --binary --fuzzy
 
 # Make will compile, download bundles and run tests

From fede2b1d714970d3a6cd5c2575bb4e4ab3799538 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 13:05:01 +0000
Subject: [PATCH 12/46] ZeroMQ version warning

---
 README.md | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/README.md b/README.md
index f2b99429..26511171 100644
--- a/README.md
+++ b/README.md
@@ -115,6 +115,11 @@ command to build Log Courier with the ZMQ transports.
 	# ZeroMQ >=4.0 - both cleartext 'plainzmq' and encrypted 'zmq' transport
 	make with=zmq4
 
+**Please ensure that the versions of ZeroMQ installed on the Logstash hosts and
+the Log Courier hosts are of the same major version. A Log Courier host that has
+ZeroMQ 4.0.5 will not work with a Logstash host using ZeroMQ 3.2.4 (but will
+work with a Logstash host using ZeroMQ 4.0.4.)**
+
 *If you receive errors whilst running `make` try `gmake` instead.*
 
 ### Generating Certificates and Keys

From 61e306fab3e84b21b05905a5e487dc86cc34d8c0 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 13:06:53 +0000
Subject: [PATCH 13/46] Cleanup moved ZMQ readme parts. Clean github.com
 downloads when make cleans

---
 Makefile  | 1 +
 README.md | 9 +++------
 2 files changed, 4 insertions(+), 6 deletions(-)

diff --git a/Makefile b/Makefile
index fee4829b..62c87022 100644
--- a/Makefile
+++ b/Makefile
@@ -84,6 +84,7 @@ vendor/bundle/.GemfileJRubyModT: Gemfile
 clean:
 	go clean -i ./...
 ifneq ($(implyclean),yes)
+	rm -rf src/github.com
 	rm -rf vendor/bundle
 	rm -f Gemfile.lock
 	rm -f log-courier-*.gem
diff --git a/README.md b/README.md
index 26511171..4a9a0770 100644
--- a/README.md
+++ b/README.md
@@ -57,15 +57,12 @@ two Logstash instances.
 1. GNU make
 
 *\*nix: Most requirements can usually be installed by your favourite package
-manager. The optional ZeroMQ >=3.2 is usually also available via the package
-manager. ZeroMQ >=4.0 may need to be build and installed manually.*
+manager.*
 
-*OS X: Git and GNU make are provided automatically by XCode. The optional ZeroMQ
-can be installed via [Homebrew](http://brew.sh).*
+*OS X: Git and GNU make are provided automatically by XCode.*
 
 *Windows: GNU make for Windows can be found
-[here](http://gnuwin32.sourceforge.net/packages/make.htm). The optional ZeroMQ
-may need to be built and installed manually.*
+[here](http://gnuwin32.sourceforge.net/packages/make.htm).*
 
 ### Building
 

From 12578857190c3a1dacaf7de35bad8e3e0c2fb9ae Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 13:08:14 +0000
Subject: [PATCH 14/46] Tidy features list RE logstash integration

---
 README.md | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/README.md b/README.md
index 4a9a0770..7c618d85 100644
--- a/README.md
+++ b/README.md
@@ -36,16 +36,15 @@ Log Courier implements the following features:
 * TLS client certificate verification
 * Secure CurveZMQ shipping transport to load balance across multiple Logstash
 instances (optional, requires ZeroMQ 4+)
-* Plaintext TCP shipping transport for configuration simplicity in local networks
+* Plaintext TCP shipping transport for configuration simplicity in local
+networks
 * Plaintext ZMQ shipping transport
 * [Administration utility](docs/AdministrationUtility.md) to monitor the
 shipping speed and status
 * [Multiline](docs/codecs/Multiline.md) codec
 * [Filter](docs/codecs/Filter.md) codec
-
-Log Courier integrates with Logstash using an event receiver ruby gem. An event
-sender ruby gem is also available to allow fast and secure transmission between
-two Logstash instances.
+* [Logstash Integration](docs/LogstashIntegration.md) with an input and output
+plugin
 
 ## Installation
 

From 1533357cfb235e33a55a6865c07b5bfe30eaa0cb Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 13:15:14 +0000
Subject: [PATCH 15/46] Improve some layout

---
 README.md | 29 ++++++++++++-----------------
 1 file changed, 12 insertions(+), 17 deletions(-)

diff --git a/README.md b/README.md
index 7c618d85..2bd5a389 100644
--- a/README.md
+++ b/README.md
@@ -55,18 +55,15 @@ plugin
 1. [git](http://git-scm.com)
 1. GNU make
 
-*\*nix: Most requirements can usually be installed by your favourite package
+* ***\*nix:*** *Most requirements can usually be installed by your favourite package
 manager.*
-
-*OS X: Git and GNU make are provided automatically by XCode.*
-
-*Windows: GNU make for Windows can be found
+* ***OS X:*** *Git and GNU make are provided automatically by XCode.*
+* ***Windows:*** *GNU make for Windows can be found
 [here](http://gnuwin32.sourceforge.net/packages/make.htm).*
 
 ### Building
 
-To build without the optional ZMQ support, simply run `make` as
-follows.
+To build, simply run `make` as follows.
 
 	git clone https://github.com/driskell/log-courier
 	cd log-courier
@@ -74,7 +71,7 @@ follows.
 
 The log-courier program can then be found in the 'bin' folder.
 
-*If you receive errors whilst running `make` try `gmake` instead.*
+*Note: If you receive errors whilst running `make`, try `gmake` instead.*
 
 ### Logstash Integration
 
@@ -92,16 +89,14 @@ on the [Logstash Integration](docs/LogstashIntegration.md) page.
 
 ### Building with ZMQ support
 
-To use the 'zmq' and 'plainzmq' transports, you will need to install
+To use the 'plainzmq' and 'zmq' transports, you will need to install
 [ZeroMQ](http://zeromq.org/intro:get-the-software) (>=3.2 for cleartext
-plainzmq, >=4.0 for encrypted zmq).
+'plainzmq', >=4.0 for encrypted 'zmq').
 
-*\*nix: ZeroMQ >=3.2 is usually available via the package manager. ZeroMQ >=4.0
+* ***\*nix:*** *ZeroMQ >=3.2 is usually available via the package manager. ZeroMQ >=4.0
 may need to be built and installed manually.*
-
-*OS X: ZeroMQ can be installed via [Homebrew](http://brew.sh).*
-
-*Windows: ZeroMQ will need to be built and installed manually.*
+* ***OS X:*** *ZeroMQ can be installed via [Homebrew](http://brew.sh).*
+* ***Windows:*** *ZeroMQ will need to be built and installed manually.*
 
 Once the required version of ZeroMQ is installed, run the corresponding `make`
 command to build Log Courier with the ZMQ transports.
@@ -111,13 +106,13 @@ command to build Log Courier with the ZMQ transports.
 	# ZeroMQ >=4.0 - both cleartext 'plainzmq' and encrypted 'zmq' transport
 	make with=zmq4
 
+*Note: If you receive errors whilst running `make`, try `gmake` instead.*
+
 **Please ensure that the versions of ZeroMQ installed on the Logstash hosts and
 the Log Courier hosts are of the same major version. A Log Courier host that has
 ZeroMQ 4.0.5 will not work with a Logstash host using ZeroMQ 3.2.4 (but will
 work with a Logstash host using ZeroMQ 4.0.4.)**
 
-*If you receive errors whilst running `make` try `gmake` instead.*
-
 ### Generating Certificates and Keys
 
 Running `make selfsigned` will automatically build and run the `lc-tlscert`

From 2b70671fdbbd566484a1e940673556915ee4c3ad Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 13:18:26 +0000
Subject: [PATCH 16/46] Fix issue in build requirements with paragraph
 appearing in GNU make

---
 README.md | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/README.md b/README.md
index 2bd5a389..5a8b6b6b 100644
--- a/README.md
+++ b/README.md
@@ -55,10 +55,10 @@ plugin
 1. [git](http://git-scm.com)
 1. GNU make
 
-* ***\*nix:*** *Most requirements can usually be installed by your favourite package
-manager.*
-* ***OS X:*** *Git and GNU make are provided automatically by XCode.*
-* ***Windows:*** *GNU make for Windows can be found
+***\*nix:*** *Most requirements can usually be installed by your favourite package
+manager.*  
+***OS X:*** *Git and GNU make are provided automatically by XCode.*  
+***Windows:*** *GNU make for Windows can be found
 [here](http://gnuwin32.sourceforge.net/packages/make.htm).*
 
 ### Building
@@ -93,10 +93,10 @@ To use the 'plainzmq' and 'zmq' transports, you will need to install
 [ZeroMQ](http://zeromq.org/intro:get-the-software) (>=3.2 for cleartext
 'plainzmq', >=4.0 for encrypted 'zmq').
 
-* ***\*nix:*** *ZeroMQ >=3.2 is usually available via the package manager. ZeroMQ >=4.0
-may need to be built and installed manually.*
-* ***OS X:*** *ZeroMQ can be installed via [Homebrew](http://brew.sh).*
-* ***Windows:*** *ZeroMQ will need to be built and installed manually.*
+***\*nix:*** *ZeroMQ >=3.2 is usually available via the package manager. ZeroMQ >=4.0
+may need to be built and installed manually.*  
+***OS X:*** *ZeroMQ can be installed via [Homebrew](http://brew.sh).*  
+***Windows:*** *ZeroMQ will need to be built and installed manually.*
 
 Once the required version of ZeroMQ is installed, run the corresponding `make`
 command to build Log Courier with the ZMQ transports.

From 82eaf8cb85c52d583b5aa98e9f5c150fbb9ec880 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 13:22:13 +0000
Subject: [PATCH 17/46] Put rvm get stable back in travis along with key pull

---
 .travis.yml | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index 479e4fba..1c43ea7c 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,9 +5,15 @@ go:
   - 1.2
   - 1.3
 
+# Until below issue is fixed, we need to install keys for rvm get stable
+# https://github.com/travis-ci/travis-ci/issues/2919
+before_install:
+  - gpg --keyserver hkp://keys.gnupg.net --recv-keys D39DC0E3 BF04FF17
+
+# Tests require ruby - ensure latest rvm list available and then switch to 2.0
 # https://github.com/travis-ci/travis-ci/issues/2220
-# Tests require ruby
 install:
+  - rvm get stable
   - rvm use 2.0 --install --binary --fuzzy
 
 # Make will compile, download bundles and run tests

From 47e8110a961655bbfadac593c72e2dc6590ffa34 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 14:56:04 +0000
Subject: [PATCH 18/46] Remove dirty flag - its not as workable as originally
 though

---
 build/fix_version | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/build/fix_version b/build/fix_version
index abccad94..d696e805 100755
--- a/build/fix_version
+++ b/build/fix_version
@@ -11,10 +11,6 @@ else
 	# Describe version from Git, and ensure the only "-xxx" is the git revision
 	# This ensures that gem builds only add one ".pre" tag automatically
 	VERSION="$(git describe | sed 's/-\([0-9][0-9]*\)-\([0-9a-z][0-9a-z]*\)$/.\1.\2/g')"
-	# Append "D" to show a dirty build, but ignore changes to versions in gemspecs etc.
-	if [ $(git status -s 2>/dev/null | grep -vE '\s(log-courier.gemspec|src/lc-lib/core/version.go)$' | wc -l) -ne 0 ]; then
-		VERSION="${VERSION}D"
-	fi
 fi
 
 # Patch version.go

From 27afe89f000e1a4336b4bd48127580a5f291737f Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 14:56:15 +0000
Subject: [PATCH 19/46] Push all gems to rubygems

---
 Rakefile | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/Rakefile b/Rakefile
index d1894959..836926c3 100644
--- a/Rakefile
+++ b/Rakefile
@@ -7,7 +7,8 @@ end
 desc 'Publish gem to RubyGems.org'
 task :publish_gem do
   require 'gem_publisher'
-  gem_file = Dir.glob(File.expand_path('*.gemspec', File.dirname(__FILE__))).first
-  gem = GemPublisher.publish_if_updated(gem_file, :rubygems)
-  puts "Published #{gem}" if gem
+  Dir.glob(File.expand_path('*.gemspec', File.dirname(__FILE__))).each do |gem_file|
+    gem = GemPublisher.publish_if_updated(gem_file, :rubygems)
+    puts "Published #{gem}" if gem
+  end
 end

From a82ca4cb87eaa8d22121709ee74af8127dfc0553 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 14:57:07 +0000
Subject: [PATCH 20/46] Make gem has no requirement to build the plugin gems as
 only the pluginmanager needs use those - this resolves problems with people
 with old gem versions which do not support "metadata=" which the plugin gems
 use

---
 Makefile | 2 --
 1 file changed, 2 deletions(-)

diff --git a/Makefile b/Makefile
index 62c87022..6e675b2f 100644
--- a/Makefile
+++ b/Makefile
@@ -38,8 +38,6 @@ log-courier: | $(BINS)
 
 gem: | fix_version
 	gem build log-courier.gemspec
-	gem build logstash-input-log-courier.gemspec
-	gem build logstash-output-log-courier.gemspec
 
 publish_gem: | fix_version vendor/bundle/.GemfileModT
 	bundle exec rake publish_gem

From b88cff6ae681912333742d4ee15254dcaa1a30b0 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 15:19:48 +0000
Subject: [PATCH 21/46] Remove rake and gem_publisher, replacing with gem push
 - we do not desire auto tag

---
 .gitignore        |  1 +
 Gemfile           |  4 ----
 Makefile          | 12 ++++++++----
 Rakefile          | 14 --------------
 build/fix_version |  3 ++-
 build/push_gems   |  8 ++++++++
 6 files changed, 19 insertions(+), 23 deletions(-)
 delete mode 100644 Rakefile
 create mode 100755 build/push_gems

diff --git a/.gitignore b/.gitignore
index 56a6fe27..84c75c4d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,6 +5,7 @@ pkg
 spec/tmp
 src/github.com
 vendor
+version.txt
 .Makefile.tags
 .bundle
 .log-courier
diff --git a/Gemfile b/Gemfile
index 4b5a27a8..dd099929 100644
--- a/Gemfile
+++ b/Gemfile
@@ -13,7 +13,3 @@ gem 'ruby-prof', :platforms => :mri
 
 # Tests
 gem 'rspec'
-
-# Publishing to ruby gems
-gem 'rake'
-gem 'gem_publisher'
diff --git a/Makefile b/Makefile
index 6e675b2f..8df171ca 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: prepare fix_version all log-courier gem test doc profile benchmark jrprofile jrbenchmark clean
+.PHONY: prepare fix_version all log-courier gem gem_plugins push_gems test doc profile benchmark jrprofile jrbenchmark clean
 
 MAKEFILE := $(word $(words $(MAKEFILE_LIST)),$(MAKEFILE_LIST))
 GOPATH := $(patsubst %/,%,$(dir $(abspath $(MAKEFILE))))
@@ -39,8 +39,12 @@ log-courier: | $(BINS)
 gem: | fix_version
 	gem build log-courier.gemspec
 
-publish_gem: | fix_version vendor/bundle/.GemfileModT
-	bundle exec rake publish_gem
+gem_plugins: | fix_version
+	gem build logstash-input-log-courier.gemspec
+	gem build logstash-output-log-courier.gemspec
+
+push_gems: | gem gem_plugins fix_version vendor/bundle/.GemfileModT
+	build/push_gems
 
 test: | all vendor/bundle/.GemfileModT
 	go get -d -tags "$(TAGS)" $(GOTESTS)
@@ -85,7 +89,7 @@ ifneq ($(implyclean),yes)
 	rm -rf src/github.com
 	rm -rf vendor/bundle
 	rm -f Gemfile.lock
-	rm -f log-courier-*.gem
+	rm -f *.gem
 endif
 
 fix_version:
diff --git a/Rakefile b/Rakefile
deleted file mode 100644
index 836926c3..00000000
--- a/Rakefile
+++ /dev/null
@@ -1,14 +0,0 @@
-task :default do
-  system('rake -T')
-end
-
-# Publish gem to rubygems
-# https://github.com/logstash-plugins/logstash-input-tcp/blob/master/rakelib/publish.rake
-desc 'Publish gem to RubyGems.org'
-task :publish_gem do
-  require 'gem_publisher'
-  Dir.glob(File.expand_path('*.gemspec', File.dirname(__FILE__))).each do |gem_file|
-    gem = GemPublisher.publish_if_updated(gem_file, :rubygems)
-    puts "Published #{gem}" if gem
-  end
-end
diff --git a/build/fix_version b/build/fix_version
index d696e805..9cc6cb05 100755
--- a/build/fix_version
+++ b/build/fix_version
@@ -27,4 +27,5 @@ for GEM in log-courier logstash-input-log-courier logstash-output-log-courier; d
 	}
 done
 
-echo "Setting Log Courier Version ${VERSION}"
+echo "${VERSION#v}" > version.txt
+echo "Set Log Courier Version ${VERSION}"
diff --git a/build/push_gems b/build/push_gems
new file mode 100755
index 00000000..5f199bea
--- /dev/null
+++ b/build/push_gems
@@ -0,0 +1,8 @@
+#!/bin/bash
+
+for GEM in *-$(cat version.txt).gem; do
+	echo "- ${GEM}"
+	gem push $GEM
+done
+
+:

From cbaf7777e39541b9d473617a8204e20997bc68b8 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 16:26:57 +0000
Subject: [PATCH 22/46] Fix partial ACK routing problem with ZMQ gem (potential
 for #63)

---
 lib/log-courier/server_zmq.rb | 19 ++++++++++++-------
 1 file changed, 12 insertions(+), 7 deletions(-)

diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index 2d869e20..2063afa7 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -91,6 +91,8 @@ def initialize(options = {})
 
       # TODO: Implement workers option by receiving on a ROUTER and proxying to a DEALER, with workers connecting to the DEALER
 
+      @return_route = []
+
       reset_timeout
     end
 
@@ -126,21 +128,18 @@ def run(&block)
             next
           end
 
-          # Pre-send the routing information and remove it from data
+          # Save the routing information and remove it from data
+          @return_route = []
           data.delete_if do |msg|
             reset_timeout
-            send_with_poll msg, true
-            if ZMQ::Util.errno != ZMQ::EAGAIN
-              @logger.warn "[LogCourierServer] Message send failed: #{ZMQ::Util.error_string}" unless @logger.nil?
-              raise TimeoutError
-            end
             break if msg == ""
+            @return_route.push msg
             true
           end
           data.shift
 
           if data.length != 1
-            @logger.warn '[LogCourierServer] Invalid message: multipart unexpected' unless @logger.nil?
+            @logger.warn "[LogCourierServer] Invalid message: multipart unexpected (#{data.length})" unless @logger.nil?
           else
             recv(data.first, &block)
           end
@@ -189,6 +188,12 @@ def recv(data)
     def send(signature, message)
       reset_timeout
       data = signature + [message.length].pack('N') + message
+
+      # Send the return route and then the message
+      @return_route.each do |msg|
+        send_with_poll msg, true
+      end
+      send_with_poll "", true
       send_with_poll data
     end
 

From eedc58dc96216299294e8200254d5afcfa335bd7 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 16:27:58 +0000
Subject: [PATCH 23/46] Fix partial ACK calculation that prevented registrar
 from persisting partially acknowledged entries

---
 src/lc-lib/publisher/publisher.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/lc-lib/publisher/publisher.go b/src/lc-lib/publisher/publisher.go
index 9d55851e..96b79eaf 100644
--- a/src/lc-lib/publisher/publisher.go
+++ b/src/lc-lib/publisher/publisher.go
@@ -547,7 +547,7 @@ func (p *Publisher) processAck(message []byte) (err error) {
     p.line_count += int64(sequence)-int64(payload.ack_events-payload.payload_start)
 
     // Only process the ACK if something was actually processed
-    if int(sequence) > payload.num_events-payload.ack_events {
+    if int(sequence) > payload.ack_events-payload.payload_start {
       payload.ack_events = int(sequence) + payload.payload_start
       // If we need to resend, we'll need to regenerate payload, so free that memory early
       payload.payload = nil

From 6670fb663d5483c997d0b1dcbe31daf3904b22c5 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 16:28:12 +0000
Subject: [PATCH 24/46] Comment tweak/elaboration

---
 src/lc-lib/transports/zmq.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/lc-lib/transports/zmq.go b/src/lc-lib/transports/zmq.go
index 55f56cc7..c082281d 100644
--- a/src/lc-lib/transports/zmq.go
+++ b/src/lc-lib/transports/zmq.go
@@ -424,7 +424,7 @@ RetryControl:
       // Try again
       goto RetryControl
     case syscall.EAGAIN:
-      // Poll lied, poll again
+      // No more messages
       return true
     }
 
@@ -532,7 +532,7 @@ RetrySend:
       // Try again
       goto RetrySend
     case syscall.EAGAIN:
-      // Poll lied, poll again
+      // No more messages
       ok = true
       return
     }
@@ -558,7 +558,7 @@ func (t *TransportZmq) processDealerIn() (ok bool) {
         // Try again
         goto RetryRecv
       case syscall.EAGAIN:
-        // Poll lied, poll again
+        // No more messages
         ok = true
         return
       }
@@ -628,7 +628,7 @@ func (t *TransportZmq) processMonitorIn() (ok bool) {
         // Try again
         goto RetryRecv
       case syscall.EAGAIN:
-        // Poll lied, poll again
+        // No more messages
         ok = true
         return
       }

From 33dcb670139b996bde67f63280b21ce245e0f06a Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 16:30:40 +0000
Subject: [PATCH 25/46] Log acknowledgements by the gem in debug mode to aid
 diagnostics

---
 lib/log-courier/server.rb | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/lib/log-courier/server.rb b/lib/log-courier/server.rb
index 5be65bb8..08df68d8 100644
--- a/lib/log-courier/server.rb
+++ b/lib/log-courier/server.rb
@@ -169,6 +169,7 @@ def process_jdat(message, comm, event_queue)
         rescue TimeoutError
           # Full pipeline, partial ack
           # NOTE: comm.send can raise a Timeout::Error of its own
+          @logger.debug "[LogCourierServer] Partially acknowledging message #{nonce.hash} sequence #{sequence}" unless @logger.nil?
           comm.send 'ACKN', [nonce, sequence].pack('A*N')
           ack_timeout = Time.now.to_i + 5
           retry
@@ -179,6 +180,7 @@ def process_jdat(message, comm, event_queue)
 
       # Acknowledge the full message
       # NOTE: comm.send can raise a Timeout::Error
+      @logger.debug "[LogCourierServer] Acknowledging message #{nonce.hash} sequence #{sequence}" unless @logger.nil?
       comm.send 'ACKN', [nonce, sequence].pack('A*N')
     end
   end

From 9a002f417442dda7797455f73b7c611483b53d90 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 16:30:52 +0000
Subject: [PATCH 26/46] @logger can be nil

---
 lib/log-courier/server_tcp.rb | 2 +-
 lib/log-courier/server_zmq.rb | 8 ++++----
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/lib/log-courier/server_tcp.rb b/lib/log-courier/server_tcp.rb
index f209cc66..caf5423f 100644
--- a/lib/log-courier/server_tcp.rb
+++ b/lib/log-courier/server_tcp.rb
@@ -100,7 +100,7 @@ def initialize(options = {})
         end
 
         if @options[:port] == 0
-          @logger.warn '[LogCourierServer] Transport ' + @options[:transport] + ' is listening on ephemeral port ' + @port.to_s
+          @logger.warn '[LogCourierServer] Transport ' + @options[:transport] + ' is listening on ephemeral port ' + @port.to_s unless @logger.nil?
         end
       rescue => e
         raise "[LogCourierServer] Failed to initialise: #{e}"
diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index 2063afa7..c587990f 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -79,15 +79,15 @@ def initialize(options = {})
         @poller = ZMQ::Poller.new
 
         if @options[:port] == 0
-          @logger.warn '[LogCourierServer] Transport ' + @options[:transport] + ' is listening on ephemeral port ' + @port.to_s
+          @logger.warn '[LogCourierServer] Transport ' + @options[:transport] + ' is listening on ephemeral port ' + @port.to_s unless @logger.nil?
         end
       rescue => e
         raise "[LogCourierServer] Failed to initialise: #{e}"
       end
 
-      @logger.info "[LogCourierServer] libzmq version #{libversion}"
-      @logger.info "[LogCourierServer] ffi-rzmq-core version #{LibZMQ::VERSION}"
-      @logger.info "[LogCourierServer] ffi-rzmq version #{ZMQ.version}"
+      @logger.info "[LogCourierServer] libzmq version #{libversion}" unless @logger.nil?
+      @logger.info "[LogCourierServer] ffi-rzmq-core version #{LibZMQ::VERSION}" unless @logger.nil?
+      @logger.info "[LogCourierServer] ffi-rzmq version #{ZMQ.version}" unless @logger.nil?
 
       # TODO: Implement workers option by receiving on a ROUTER and proxying to a DEALER, with workers connecting to the DEALER
 

From d17f84beb438f84657e002c48fafbf503a697be9 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 19:15:38 +0000
Subject: [PATCH 27/46] Fix up event monitoring for zeromq 3.2

---
 src/lc-lib/transports/zmq.go         | 79 +++++++---------------------
 src/lc-lib/transports/zmq_curve.go   | 60 +++++++++++++++++++++
 src/lc-lib/transports/zmq_nocurve.go | 78 +++++++++++++++++++++++++++
 3 files changed, 158 insertions(+), 59 deletions(-)

diff --git a/src/lc-lib/transports/zmq.go b/src/lc-lib/transports/zmq.go
index c082281d..9e59f58c 100644
--- a/src/lc-lib/transports/zmq.go
+++ b/src/lc-lib/transports/zmq.go
@@ -87,17 +87,33 @@ type ZMQEvent struct {
   part  int
   event zmq.Event
   val   int32
-  data  []byte
+  data  string
 }
 
 func (e *ZMQEvent) Log() {
   switch e.event {
   case zmq.EVENT_CONNECTED:
-    log.Info("Connected to %s", e.data)
+    if e.data == "" {
+      log.Info("Connected")
+    } else {
+      log.Info("Connected to %s", e.data)
+    }
+  case zmq.EVENT_CONNECT_DELAYED:
+    // Don't log anything for this
   case zmq.EVENT_CONNECT_RETRIED:
-    log.Info("Attempting to connect to %s", e.data)
+    if e.data == "" {
+      log.Info("Attempting to connect")
+    } else {
+      log.Info("Attempting to connect to %s", e.data)
+    }
   case zmq.EVENT_DISCONNECTED:
-    log.Error("Lost connection to %s", e.data)
+    if e.data == "" {
+      log.Error("Lost connection")
+    } else {
+      log.Error("Lost connection to %s", e.data)
+    }
+  default:
+    log.Debug("Unknown monitor message (event:%d, val:%d, data:[% X])", e.event, e.val, e.data)
   }
 }
 
@@ -617,61 +633,6 @@ func (t *TransportZmq) processDealerIn() (ok bool) {
   }
 }
 
-func (t *TransportZmq) processMonitorIn() (ok bool) {
-  for {
-    // Bring in the messages
-  RetryRecv:
-    data, err := t.monitor.Recv(zmq.DONTWAIT)
-    if err != nil {
-      switch err {
-      case syscall.EINTR:
-        // Try again
-        goto RetryRecv
-      case syscall.EAGAIN:
-        // No more messages
-        ok = true
-        return
-      }
-
-      // Failure
-      t.recv_chan <- fmt.Errorf("Monitor zmq.Socket.Recv failure %s", err)
-      return
-    }
-
-    more, err := t.monitor.RcvMore()
-    if err != nil {
-      // Failure
-      t.recv_chan <- fmt.Errorf("Monitor zmq.Socket.RcvMore failure %s", err)
-      return
-    }
-
-    switch t.event.part {
-    case Monitor_Part_Header:
-      t.event.event = zmq.Event(binary.LittleEndian.Uint16(data[0:2]))
-      t.event.val = int32(binary.LittleEndian.Uint32(data[2:6]))
-    case Monitor_Part_Data:
-      t.event.data = data
-      t.event.Log()
-    default:
-      log.Debug("Extraneous data in monitor message. Silently discarding.")
-      continue
-    }
-
-    if !more {
-      if t.event.part < Monitor_Part_Data {
-        log.Debug("Unexpected end of monitor message. Skipping.")
-      }
-
-      t.event.part = Monitor_Part_Header
-      continue
-    }
-
-    if t.event.part <= Monitor_Part_Data {
-      t.event.part++
-    }
-  }
-}
-
 func (t *TransportZmq) setChan(set chan int) {
   select {
   case set <- 1:
diff --git a/src/lc-lib/transports/zmq_curve.go b/src/lc-lib/transports/zmq_curve.go
index a7c03315..eb07a835 100644
--- a/src/lc-lib/transports/zmq_curve.go
+++ b/src/lc-lib/transports/zmq_curve.go
@@ -20,6 +20,7 @@ package transports
 
 import (
   "fmt"
+  zmq "github.com/alecthomas/gozmq"
   "lc-lib/core"
 )
 
@@ -59,6 +60,65 @@ func (t *TransportZmq) configureSocket() (err error) {
   return
 }
 
+// Process ZMQ 4.0.x monitor messages
+// http://api.zeromq.org/4-0:zmq-socket-monitor
+func (t *TransportZmq) processMonitorIn() (ok bool) {
+  for {
+    // Bring in the messages
+  RetryRecv:
+    data, err := t.monitor.Recv(zmq.DONTWAIT)
+    if err != nil {
+      switch err {
+      case syscall.EINTR:
+        // Try again
+        goto RetryRecv
+      case syscall.EAGAIN:
+        // No more messages
+        ok = true
+        return
+      }
+
+      // Failure
+      t.recv_chan <- fmt.Errorf("Monitor zmq.Socket.Recv failure %s", err)
+      return
+    }
+
+    switch t.event.part {
+    case Monitor_Part_Header:
+      t.event.event = zmq.Event(binary.LittleEndian.Uint16(data[0:2]))
+      t.event.val = int32(binary.LittleEndian.Uint32(data[2:6]))
+      t.event.data = ""
+    case Monitor_Part_Data:
+      t.event.data = string(data)
+      t.event.Log()
+    default:
+      log.Debug("Extraneous data in monitor message. Silently discarding.")
+      continue
+    }
+
+    more, err := t.monitor.RcvMore()
+    if err != nil {
+      // Failure
+      t.recv_chan <- fmt.Errorf("Monitor zmq.Socket.RcvMore failure %s", err)
+      return
+    }
+
+    if !more {
+      if t.event.part < Monitor_Part_Data {
+        t.event.Log()
+        log.Debug("Unexpected end of monitor message. Skipping.")
+      }
+
+      t.event.part = Monitor_Part_Header
+      continue
+    }
+
+    if t.event.part <= Monitor_Part_Data {
+      t.event.part++
+    }
+  }
+}
+
 // Register the transport
 func init() {
   core.RegisterTransport("zmq", NewZmqTransportFactory)
diff --git a/src/lc-lib/transports/zmq_nocurve.go b/src/lc-lib/transports/zmq_nocurve.go
index ce39bb07..f131b07e 100644
--- a/src/lc-lib/transports/zmq_nocurve.go
+++ b/src/lc-lib/transports/zmq_nocurve.go
@@ -18,6 +18,25 @@
 
 package transports
 
+/*
+#cgo pkg-config: libzmq
+#include <zmq.h>
+
+struct zmq_event_t_wrap {
+  int event;
+  char *addr;
+  int fd;
+};
+*/
+import "C"
+
+import (
+  "fmt"
+  zmq "github.com/alecthomas/gozmq"
+  "syscall"
+  "unsafe"
+)
+
 func (f *TransportZmqFactory) processConfig(config_path string) (err error) {
   return nil
 }
@@ -25,3 +44,62 @@ func (f *TransportZmqFactory) processConfig(config_path string) (err error) {
 func (t *TransportZmq) configureSocket() error {
   return nil
 }
+
+// Process ZMQ 3.2.x monitor messages
+// http://api.zeromq.org/3-2:zmq-socket-monitor
+func (t *TransportZmq) processMonitorIn() (ok bool) {
+  for {
+    // Bring in the messages
+  RetryRecv:
+    data, err := t.monitor.Recv(zmq.DONTWAIT)
+    if err != nil {
+      switch err {
+      case syscall.EINTR:
+        // Try again
+        goto RetryRecv
+      case syscall.EAGAIN:
+        // No more messages
+        ok = true
+        return
+      }
+
+      // Failure
+      t.recv_chan <- fmt.Errorf("Monitor zmq.Socket.Recv failure %s", err)
+      return
+    }
+
+    switch t.event.part {
+    case Monitor_Part_Header:
+      event := (*C.struct_zmq_event_t_wrap)(unsafe.Pointer(&data[0]))
+      t.event.event = zmq.Event(event.event)
+      if event.addr == nil {
+        t.event.data = ""
+      } else {
+        // TODO: Fix this - data has been feed by zmq_msg_close!
+        //t.event.data = C.GoString(event.addr)
+        t.event.data = ""
+      }
+      t.event.val = int32(event.fd)
+      t.event.Log()
+    default:
+      log.Debug("Extraneous data in monitor message. Silently discarding.")
+      continue
+    }
+
+    more, err := t.monitor.RcvMore()
+    if err != nil {
+      // Failure
+      t.recv_chan <- fmt.Errorf("Monitor zmq.Socket.RcvMore failure %s", err)
+      return
+    }
+
+    if !more {
+      t.event.part = Monitor_Part_Header
+      continue
+    }
+
+    if t.event.part <= Monitor_Part_Data {
+      t.event.part++
+    }
+  }
+}

From b96281bd507d29ba0969160529515281c24da271 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Wed, 29 Oct 2014 19:22:22 +0000
Subject: [PATCH 28/46] Tidy zmq naming and fix for 4.x

---
 .../transports/{zmq_nocurve.go => zmq3.go}    |  0
 .../transports/{zmq_curve.go => zmq4.go}      | 32 ++++++++++++
 src/lc-lib/transports/zmq_z85validate.go      | 52 -------------------
 3 files changed, 32 insertions(+), 52 deletions(-)
 rename src/lc-lib/transports/{zmq_nocurve.go => zmq3.go} (100%)
 rename src/lc-lib/transports/{zmq_curve.go => zmq4.go} (86%)
 delete mode 100644 src/lc-lib/transports/zmq_z85validate.go

diff --git a/src/lc-lib/transports/zmq_nocurve.go b/src/lc-lib/transports/zmq3.go
similarity index 100%
rename from src/lc-lib/transports/zmq_nocurve.go
rename to src/lc-lib/transports/zmq3.go
diff --git a/src/lc-lib/transports/zmq_curve.go b/src/lc-lib/transports/zmq4.go
similarity index 86%
rename from src/lc-lib/transports/zmq_curve.go
rename to src/lc-lib/transports/zmq4.go
index eb07a835..4b783292 100644
--- a/src/lc-lib/transports/zmq_curve.go
+++ b/src/lc-lib/transports/zmq4.go
@@ -18,10 +18,20 @@
 
 package transports
 
+/*
+#cgo pkg-config: libzmq
+#include <zmq.h>
+#include <zmq_utils.h>
+*/
+import "C"
+
 import (
+  "encoding/binary"
   "fmt"
   zmq "github.com/alecthomas/gozmq"
   "lc-lib/core"
+  "syscall"
+  "unsafe"
 )
 
 func (f *TransportZmqFactory) processConfig(config_path string) (err error) {
@@ -119,6 +129,28 @@ func (t *TransportZmq) processMonitorIn() (ok bool) {
   }
 }
 
+func z85Validate(z85 string) bool {
+  var decoded []C.uint8_t
+
+  if len(z85)%5 != 0 {
+    return false
+  } else {
+    // Avoid literal floats
+    decoded = make([]C.uint8_t, 8*len(z85)/10)
+  }
+
+  // Grab a CString of the z85 we need to decode
+  c_z85 := C.CString(z85)
+  defer C.free(unsafe.Pointer(c_z85))
+
+  // Because gozmq does not yet expose this for us, we have to expose it ourselves
+  if ret := C.zmq_z85_decode(&decoded[0], c_z85); ret == nil {
+    return false
+  }
+
+  return true
+}
+
 // Register the transport
 func init() {
   core.RegisterTransport("zmq", NewZmqTransportFactory)
diff --git a/src/lc-lib/transports/zmq_z85validate.go b/src/lc-lib/transports/zmq_z85validate.go
deleted file mode 100644
index 674b7618..00000000
--- a/src/lc-lib/transports/zmq_z85validate.go
+++ /dev/null
@@ -1,52 +0,0 @@
-// +build zmq_4_x
-
-/*
- * Copyright 2014 Jason Woods.
- *
- * 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.
- */
-
-package transports
-
-import (
-  "unsafe"
-)
-
-/*
-#cgo pkg-config: libzmq
-#include <zmq.h>
-#include <zmq_utils.h>
-*/
-import "C"
-
-func z85Validate(z85 string) bool {
-  var decoded []C.uint8_t
-
-  if len(z85)%5 != 0 {
-    return false
-  } else {
-    // Avoid literal floats
-    decoded = make([]C.uint8_t, 8*len(z85)/10)
-  }
-
-  // Grab a CString of the z85 we need to decode
-  c_z85 := C.CString(z85)
-  defer C.free(unsafe.Pointer(c_z85))
-
-  // Because gozmq does not yet expose this for us, we have to expose it ourselves
-  if ret := C.zmq_z85_decode(&decoded[0], c_z85); ret == nil {
-    return false
-  }
-
-  return true
-}

From d3e315c8973d5d381152931f5cdd6a1547bc1f84 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 09:29:11 +0000
Subject: [PATCH 29/46] Extra checks and logging on ZMQ message receive in gem

---
 lib/log-courier/server_zmq.rb | 19 ++++++++++++++++++-
 1 file changed, 18 insertions(+), 1 deletion(-)

diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index c587990f..37287a14 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -128,7 +128,7 @@ def run(&block)
             next
           end
 
-          # Save the routing information and remove it from data
+          # Save the routing information that appears before the null messages
           @return_route = []
           data.delete_if do |msg|
             reset_timeout
@@ -136,10 +136,27 @@ def run(&block)
             @return_route.push msg
             true
           end
+
+          if data.length == 0
+            @logger.warn '[LogCourierServer] Invalid message: no data' unless @logger.nil?
+            next
+          elsif data.length == 1
+            @logger.warn '[LogCourierServer] Invalid message: empty data' unless @logger.nil?
+            next
+          end
+
+          # Drop the null message separator
           data.shift
 
           if data.length != 1
             @logger.warn "[LogCourierServer] Invalid message: multipart unexpected (#{data.length})" unless @logger.nil?
+            if !@logger.nil? && @logger.debug?
+              i = 0
+              data.each do |msg|
+                i += 1
+                @logger.debug "[LogCourierServer] Part #{i}: #{msg[0..31].gsub(/[^[:print:]]/,'.')}"
+              end
+            end
           else
             recv(data.first, &block)
           end

From ad448f583dbbd6edbfa2df7df26fa209ce3e7a9b Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 12:42:56 +0000
Subject: [PATCH 30/46] End stdin harvester on EOF

---
 src/lc-lib/harvester/harvester.go | 21 +++++++++++++--------
 1 file changed, 13 insertions(+), 8 deletions(-)

diff --git a/src/lc-lib/harvester/harvester.go b/src/lc-lib/harvester/harvester.go
index ebc5b2e6..ac4517dc 100644
--- a/src/lc-lib/harvester/harvester.go
+++ b/src/lc-lib/harvester/harvester.go
@@ -123,7 +123,7 @@ func (h *Harvester) harvest(output chan<- *core.EventDescriptor) (int64, error)
   h.output = output
 
   if h.path == "-" {
-    log.Info("Started stdin harvester at position 0")
+    log.Info("Started stdin harvester")
     h.offset = 0
   } else {
     // Get current offset in file
@@ -200,10 +200,21 @@ ReadLoop:
     }
 
     if err != io.EOF {
-      log.Error("Unexpected error reading from %s: %s", h.path, err)
+      if h.path == "-" {
+        log.Error("Unexpected error reading from stdin: %s", err)
+      } else {
+        log.Error("Unexpected error reading from %s: %s", h.path, err)
+      }
       return h.codec.Teardown(), err
     }
 
+    if h.path == "-" {
+      // Stdin has finished - stdin blocks permanently until the stream ends
+      // Once the stream ends, finish the harvester
+      log.Info("Stopping harvest of stdin; EOF reached")
+      return h.codec.Teardown(), nil
+    }
+
     // Check shutdown
     select {
     case <-h.stop_chan:
@@ -216,12 +227,6 @@ ReadLoop:
     h.last_eof = &last_eof
     h.Unlock()
 
-    // Timed out waiting for data, got EOF
-    if h.path == "-" {
-      // This wouldn't make sense on stdin so lets not risk anything strange happening
-      continue
-    }
-
     // Don't check for truncation until we hit the full read_timeout
     if time.Since(last_read_time) < read_timeout {
       continue

From c3275f82598a133d59ce4d06d82befb6b8773936 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 12:51:31 +0000
Subject: [PATCH 31/46] Replace an rspec test with a go test

---
 spec/courier_spec.rb                    | 20 --------------------
 src/lc-lib/harvester/linereader_test.go | 11 +++++++++++
 2 files changed, 11 insertions(+), 20 deletions(-)

diff --git a/spec/courier_spec.rb b/spec/courier_spec.rb
index 07c0a65f..f93aba7b 100644
--- a/spec/courier_spec.rb
+++ b/spec/courier_spec.rb
@@ -183,26 +183,6 @@
     receive_and_check
   end
 
-  it 'should handle incomplete lines in buffered logs by waiting for a line end' do
-    f = create_log
-
-    startup
-
-    1_000.times do |i|
-      if (i + 100) % 500 == 0
-        # Make 2 events where we pause for >10s before adding new line, this takes us past eof_timeout
-        f.log_partial_start
-        sleep 15
-        f.log_partial_end
-      else
-        f.log
-      end
-    end
-
-    # Receive and check
-    receive_and_check
-  end
-
   it 'should handle log rotation and resume correctly' do
     f1 = create_log
 
diff --git a/src/lc-lib/harvester/linereader_test.go b/src/lc-lib/harvester/linereader_test.go
index aae1b030..474ca735 100644
--- a/src/lc-lib/harvester/linereader_test.go
+++ b/src/lc-lib/harvester/linereader_test.go
@@ -87,6 +87,17 @@ func TestLineReadEmpty(t *testing.T) {
   checkEnd(t, reader)
 }
 
+func TestLineReadIncomplete(t *testing.T) {
+  data := bytes.NewBufferString("\n12345678901234567890\n123456")
+
+  // New line read with 100 bytes, enough for the above
+  reader := NewLineReader(data, 100)
+
+  checkLine(t, reader, []byte("\n"))
+  checkLine(t, reader, []byte("12345678901234567890\n"))
+  checkEnd(t, reader)
+}
+
 func TestLineReadFull(t *testing.T) {
   data := bytes.NewBufferString("12345678901234567890\n123456789012345678901234567890\n12345678901234567890\n")
 

From d46faa38508c3bce2f1960d06337efb59b3e6207 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 13:00:06 +0000
Subject: [PATCH 32/46] Replace multiline rspec tests with go tests

---
 spec/multiline_spec.rb              | 176 ----------------------------
 src/lc-lib/codecs/multiline_test.go | 122 +++++++++++++++++--
 2 files changed, 114 insertions(+), 184 deletions(-)
 delete mode 100644 spec/multiline_spec.rb

diff --git a/spec/multiline_spec.rb b/spec/multiline_spec.rb
deleted file mode 100644
index dccdd4d3..00000000
--- a/spec/multiline_spec.rb
+++ /dev/null
@@ -1,176 +0,0 @@
-# encoding: utf-8
-
-# Copyright 2014 Jason Woods.
-#
-# 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 'lib/common'
-require 'lib/helpers/log-courier'
-require 'lib/logfile/multiline'
-
-describe 'log-courier with multiline codec' do
-  include_context 'Helpers'
-  include_context 'Helpers_Log_Courier'
-
-  it 'should combine multiple events with what=previous' do
-    startup config: <<-config
-    {
-      "network": {
-        "ssl ca": "#{@ssl_cert.path}",
-        "servers": [ "127.0.0.1:#{server_port}" ],
-        "timeout": 15,
-        "reconnect": 1
-      },
-      "files": [
-        {
-          "paths": [ "#{TEMP_PATH}/logs/log-*" ],
-          "codec": { "name": "multiline", "what": "previous", "pattern": "^\\\\s" }
-        }
-      ]
-    }
-    config
-
-    f = create_log(LogFile::Multiline)
-
-    5000.times do |i|
-      f.log
-    end
-
-    # We will always be missing the last line - this is expected behaviour as we cannot know the last multiline block is complete
-    f.skip_one
-
-    # Receive and check
-    receive_and_check
-  end
-
-  it 'should combine multiple events with what=previous and negate' do
-    startup config: <<-config
-    {
-      "network": {
-        "ssl ca": "#{@ssl_cert.path}",
-        "servers": [ "127.0.0.1:#{server_port}" ],
-        "timeout": 15,
-        "reconnect": 1
-      },
-      "files": [
-        {
-          "paths": [ "#{TEMP_PATH}/logs/log-*" ],
-          "codec": { "name": "multiline", "what": "previous", "pattern": "^BEGIN", "negate": true }
-        }
-      ]
-    }
-    config
-
-    f = create_log(LogFile::Multiline)
-
-    5_000.times do
-      f.log
-    end
-
-    # We will always be missing the last line - this is expected behaviour as we cannot know the last multiline block is complete
-    f.skip_one
-
-    # Receive and check
-    receive_and_check
-  end
-
-  it 'should combine multiple events with what=previous and previous_timeout' do
-    startup config: <<-config
-    {
-      "network": {
-        "ssl ca": "#{@ssl_cert.path}",
-        "servers": [ "127.0.0.1:#{server_port}" ],
-        "timeout": 15,
-        "reconnect": 1
-      },
-      "files": [
-        {
-          "paths": [ "#{TEMP_PATH}/logs/log-*" ],
-          "codec": { "name": "multiline", "what": "previous", "pattern": "^\\\\s", "previous timeout": "3s" }
-        }
-      ]
-    }
-    config
-
-    f = create_log(LogFile::Multiline)
-
-    1_500.times do
-      f.log
-    end
-
-    sleep 15
-
-    1_500.times do
-      f.log
-    end
-
-    # Receive and check
-    receive_and_check
-  end
-
-  it 'should combine multiple events with what=next' do
-    startup config: <<-config
-    {
-      "network": {
-        "ssl ca": "#{@ssl_cert.path}",
-        "servers": [ "127.0.0.1:#{server_port}" ],
-        "timeout": 15,
-        "reconnect": 1
-      },
-      "files": [
-        {
-          "paths": [ "#{TEMP_PATH}/logs/log-*" ],
-          "codec": { "name": "multiline", "what": "next", "pattern": "[0-9]$" }
-        }
-      ]
-    }
-    config
-
-    f = create_log(LogFile::Multiline)
-
-    5_000.times do
-      f.log
-    end
-
-    # Receive and check
-    receive_and_check
-  end
-
-  it 'should combine multiple events with what=next and negate' do
-    startup config: <<-config
-    {
-      "network": {
-        "ssl ca": "#{@ssl_cert.path}",
-        "servers": [ "127.0.0.1:#{server_port}" ],
-        "timeout": 15,
-        "reconnect": 1
-      },
-      "files": [
-        {
-          "paths": [ "#{TEMP_PATH}/logs/log-*" ],
-          "codec": { "name": "multiline", "what": "next", "pattern": "^\\\\sEND", "negate": true }
-        }
-      ]
-    }
-    config
-
-    f = create_log(LogFile::Multiline)
-
-    5_000.times do
-      f.log
-    end
-
-    # Receive and check
-    receive_and_check
-  end
-end
diff --git a/src/lc-lib/codecs/multiline_test.go b/src/lc-lib/codecs/multiline_test.go
index 00119dd5..2a7ef662 100644
--- a/src/lc-lib/codecs/multiline_test.go
+++ b/src/lc-lib/codecs/multiline_test.go
@@ -3,6 +3,7 @@ package codecs
 import (
   "lc-lib/core"
   "testing"
+  "time"
 )
 
 var gt *testing.T
@@ -25,30 +26,135 @@ func createCodec(unused map[string]interface{}, callback core.CodecCallbackFunc,
 func checkMultiline(start_offset int64, end_offset int64, text string) {
   lines++
 
-  if text != "DEBUG First line\nsecond line\nthird line" {
-    gt.Logf("Event data incorrect [% X]", text)
-    gt.FailNow()
+  if lines == 1 {
+    if text != "DEBUG First line\nNEXT line\nANOTHER line" {
+      gt.Logf("Event data incorrect [% X]", text)
+      gt.FailNow()
+    }
+
+    if end_offset != 5 {
+      gt.Logf("Event end offset is incorrect [%d]", end_offset)
+      gt.FailNow()
+    }
+  } else {
+    if text != "DEBUG Next line" {
+      gt.Logf("Event data incorrect [% X]", text)
+      gt.FailNow()
+    }
+
+    if end_offset != 7 {
+      gt.Logf("Event end offset is incorrect [%d]", end_offset)
+      gt.FailNow()
+    }
   }
+}
+
+func TestMultilinePrevious(t *testing.T) {
+  gt = t
+  lines = 0
+
+  codec := createCodec(map[string]interface{}{
+    "pattern": "^(ANOTHER|NEXT) ",
+    "what": "previous",
+    "negate": false,
+  }, checkMultiline, t)
+
+  // Send some data
+  codec.Event(0, 1, "DEBUG First line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
+  codec.Event(6, 7, "DEBUG Next line")
 
-  if end_offset != 5 {
-    gt.Logf("Event end offset is incorrect [%d]", end_offset)
+  if lines != 1 {
+    gt.Logf("Wrong line count received")
     gt.FailNow()
   }
 }
 
-func TestMultiline(t *testing.T) {
+func TestMultilinePreviousNegate(t *testing.T) {
   gt = t
   lines = 0
 
   codec := createCodec(map[string]interface{}{
     "pattern": "^DEBUG ",
+    "what": "previous",
     "negate": true,
   }, checkMultiline, t)
 
   // Send some data
   codec.Event(0, 1, "DEBUG First line")
-  codec.Event(2, 3, "second line")
-  codec.Event(4, 5, "third line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
+  codec.Event(6, 7, "DEBUG Next line")
+
+  if lines != 1 {
+    gt.Logf("Wrong line count received")
+    gt.FailNow()
+  }
+}
+
+func TestMultilinePreviousTimeout(t *testing.T) {
+  gt = t
+  lines = 0
+
+  codec := createCodec(map[string]interface{}{
+    "pattern": "^(ANOTHER|NEXT) ",
+    "what": "previous",
+    "negate": false,
+    "previous timeout": "5s",
+  }, checkMultiline, t)
+
+  // Send some data
+  codec.Event(0, 1, "DEBUG First line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
+  codec.Event(6, 7, "DEBUG Next line")
+
+  // Allow 7 seconds
+  time.Sleep(7 * time.Second)
+
+  if lines != 2 {
+    gt.Logf("Wrong line count received")
+    gt.FailNow()
+  }
+}
+
+func TestMultilineNext(t *testing.T) {
+  gt = t
+  lines = 0
+
+  codec := createCodec(map[string]interface{}{
+    "pattern": "^(DEBUG|NEXT) ",
+    "what": "next",
+    "negate": false,
+  }, checkMultiline, t)
+
+  // Send some data
+  codec.Event(0, 1, "DEBUG First line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
+  codec.Event(6, 7, "DEBUG Next line")
+
+  if lines != 1 {
+    gt.Logf("Wrong line count received")
+    gt.FailNow()
+  }
+}
+
+func TestMultilineNextNegate(t *testing.T) {
+  gt = t
+  lines = 0
+
+  codec := createCodec(map[string]interface{}{
+    "pattern": "^ANOTHER ",
+    "what": "next",
+    "negate": true,
+  }, checkMultiline, t)
+
+  // Send some data
+  codec.Event(0, 1, "DEBUG First line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
   codec.Event(6, 7, "DEBUG Next line")
 
   if lines != 1 {

From 6bf8068d22608b025d0c6aacf3b89834ef38cc6a Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 13:15:12 +0000
Subject: [PATCH 33/46] Replace filter codec rspec test with go test (and tweak
 up multiline codec test)

---
 spec/filter_spec.rb                 |  57 ------------
 src/lc-lib/codecs/filter_test.go    | 102 +++++++++++++++++++++
 src/lc-lib/codecs/multiline_test.go | 133 +++++++++++++++-------------
 3 files changed, 175 insertions(+), 117 deletions(-)
 delete mode 100644 spec/filter_spec.rb
 create mode 100644 src/lc-lib/codecs/filter_test.go

diff --git a/spec/filter_spec.rb b/spec/filter_spec.rb
deleted file mode 100644
index 84ba0f2c..00000000
--- a/spec/filter_spec.rb
+++ /dev/null
@@ -1,57 +0,0 @@
-# encoding: utf-8
-
-# Copyright 2014 Jason Woods.
-#
-# 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 'lib/common'
-require 'lib/helpers/log-courier'
-
-describe 'log-courier with filter codec' do
-  include_context 'Helpers'
-  include_context 'Helpers_Log_Courier'
-
-  it 'should filter events' do
-    startup stdin: true, config: <<-config
-    {
-      "network": {
-        "ssl ca": "#{@ssl_cert.path}",
-        "servers": [ "127.0.0.1:#{server_port}" ],
-        "timeout": 15,
-        "reconnect": 1
-      },
-      "files": [
-        {
-          "paths": [ "-" ],
-          "codec": { "name": "filter", "patterns": [ "^stdin line test [12]" ], "negate": true }
-        }
-      ]
-    }
-    config
-
-    5_000.times do |i|
-      @log_courier.puts "stdin line test #{i}"
-    end
-
-    # Receive and check
-    i = 0
-    host = Socket.gethostname
-    receive_and_check(total: 2_778) do |e|
-      expect(e['message']).to eq "stdin line test #{i}"
-      expect(e['host']).to eq host
-      expect(e['path']).to eq '-'
-      i += 1
-      i += 1 while /^[12]/ =~ i.to_s
-    end
-  end
-end
diff --git a/src/lc-lib/codecs/filter_test.go b/src/lc-lib/codecs/filter_test.go
new file mode 100644
index 00000000..4d34624a
--- /dev/null
+++ b/src/lc-lib/codecs/filter_test.go
@@ -0,0 +1,102 @@
+package codecs
+
+import (
+  "lc-lib/core"
+  "testing"
+)
+
+var filter_lines []string
+
+func createFilterCodec(unused map[string]interface{}, callback core.CodecCallbackFunc, t *testing.T) core.Codec {
+  config := core.NewConfig()
+
+  factory, err := NewFilterCodecFactory(config, "", unused, "filter")
+  if err != nil {
+    t.Logf("Failed to create filter codec: %s", err)
+    t.FailNow()
+  }
+
+  return factory.NewCodec(callback, 0)
+}
+
+func checkFilter(start_offset int64, end_offset int64, text string) {
+  filter_lines = append(filter_lines, text)
+}
+
+func TestFilter(t *testing.T) {
+  filter_lines = make([]string, 0, 1)
+
+  codec := createFilterCodec(map[string]interface{}{
+    "patterns": []string{"^NEXT line$"},
+    "negate": false,
+  }, checkFilter, t)
+
+  // Send some data
+  codec.Event(0, 1, "DEBUG First line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
+  codec.Event(6, 7, "DEBUG Next line")
+
+  if len(filter_lines) != 1 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
+  } else if filter_lines[0] != "NEXT line" {
+    t.Logf("Wrong line[0] received: %s", filter_lines[0])
+    t.FailNow()
+  }
+}
+
+func TestFilterNegate(t *testing.T) {
+  filter_lines = make([]string, 0, 1)
+
+  codec := createFilterCodec(map[string]interface{}{
+    "patterns": []string{"^NEXT line$"},
+    "negate": true,
+  }, checkFilter, t)
+
+  // Send some data
+  codec.Event(0, 1, "DEBUG First line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
+  codec.Event(6, 7, "DEBUG Next line")
+
+  if len(filter_lines) != 3 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
+  } else if filter_lines[0] != "DEBUG First line" {
+    t.Logf("Wrong line[0] received: %s", filter_lines[0])
+    t.FailNow()
+  } else if filter_lines[1] != "ANOTHER line" {
+    t.Logf("Wrong line[1] received: %s", filter_lines[1])
+    t.FailNow()
+  } else if filter_lines[2] != "DEBUG Next line" {
+    t.Logf("Wrong line[2] received: %s", filter_lines[2])
+    t.FailNow()
+  }
+}
+
+func TestFilterMultiple(t *testing.T) {
+  filter_lines = make([]string, 0, 1)
+
+  codec := createFilterCodec(map[string]interface{}{
+    "patterns": []string{"^NEXT line$", "^DEBUG First line$"},
+    "negate": false,
+  }, checkFilter, t)
+
+  // Send some data
+  codec.Event(0, 1, "DEBUG First line")
+  codec.Event(2, 3, "NEXT line")
+  codec.Event(4, 5, "ANOTHER line")
+  codec.Event(6, 7, "DEBUG Next line")
+
+  if len(filter_lines) != 2 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
+  } else if filter_lines[0] != "DEBUG First line" {
+    t.Logf("Wrong line[0] received: %s", filter_lines[0])
+    t.FailNow()
+  } else if filter_lines[1] != "NEXT line" {
+    t.Logf("Wrong line[1] received: %s", filter_lines[1])
+    t.FailNow()
+  }
+}
diff --git a/src/lc-lib/codecs/multiline_test.go b/src/lc-lib/codecs/multiline_test.go
index 2a7ef662..56b48b4b 100644
--- a/src/lc-lib/codecs/multiline_test.go
+++ b/src/lc-lib/codecs/multiline_test.go
@@ -6,10 +6,10 @@ import (
   "time"
 )
 
-var gt *testing.T
-var lines int
+var multiline_t *testing.T
+var multiline_lines int
 
-func createCodec(unused map[string]interface{}, callback core.CodecCallbackFunc, t *testing.T) core.Codec {
+func createMultilineCodec(unused map[string]interface{}, callback core.CodecCallbackFunc, t *testing.T) core.Codec {
   config := core.NewConfig()
   config.General.MaxLineBytes = 1048576
   config.General.SpoolMaxBytes = 10485760
@@ -24,36 +24,48 @@ func createCodec(unused map[string]interface{}, callback core.CodecCallbackFunc,
 }
 
 func checkMultiline(start_offset int64, end_offset int64, text string) {
-  lines++
+  multiline_lines++
 
-  if lines == 1 {
+  if multiline_lines == 1 {
     if text != "DEBUG First line\nNEXT line\nANOTHER line" {
-      gt.Logf("Event data incorrect [% X]", text)
-      gt.FailNow()
+      multiline_t.Logf("Event data incorrect [% X]", text)
+      multiline_t.FailNow()
     }
 
-    if end_offset != 5 {
-      gt.Logf("Event end offset is incorrect [%d]", end_offset)
-      gt.FailNow()
-    }
-  } else {
-    if text != "DEBUG Next line" {
-      gt.Logf("Event data incorrect [% X]", text)
-      gt.FailNow()
+    if start_offset != 0 {
+      multiline_t.Logf("Event start offset is incorrect [%d]", start_offset)
+      multiline_t.FailNow()
     }
 
-    if end_offset != 7 {
-      gt.Logf("Event end offset is incorrect [%d]", end_offset)
-      gt.FailNow()
+    if end_offset != 5 {
+      multiline_t.Logf("Event end offset is incorrect [%d]", end_offset)
+      multiline_t.FailNow()
     }
+
+    return
+  }
+
+  if text != "DEBUG Next line" {
+    multiline_t.Logf("Event data incorrect [% X]", text)
+    multiline_t.FailNow()
+  }
+
+  if start_offset != 6 {
+    multiline_t.Logf("Event start offset is incorrect [%d]", start_offset)
+    multiline_t.FailNow()
+  }
+
+  if end_offset != 7 {
+    multiline_t.Logf("Event end offset is incorrect [%d]", end_offset)
+    multiline_t.FailNow()
   }
 }
 
 func TestMultilinePrevious(t *testing.T) {
-  gt = t
-  lines = 0
+  multiline_t = t
+  multiline_lines = 0
 
-  codec := createCodec(map[string]interface{}{
+  codec := createMultilineCodec(map[string]interface{}{
     "pattern": "^(ANOTHER|NEXT) ",
     "what": "previous",
     "negate": false,
@@ -65,17 +77,17 @@ func TestMultilinePrevious(t *testing.T) {
   codec.Event(4, 5, "ANOTHER line")
   codec.Event(6, 7, "DEBUG Next line")
 
-  if lines != 1 {
-    gt.Logf("Wrong line count received")
-    gt.FailNow()
+  if multiline_lines != 1 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
   }
 }
 
 func TestMultilinePreviousNegate(t *testing.T) {
-  gt = t
-  lines = 0
+  multiline_t = t
+  multiline_lines = 0
 
-  codec := createCodec(map[string]interface{}{
+  codec := createMultilineCodec(map[string]interface{}{
     "pattern": "^DEBUG ",
     "what": "previous",
     "negate": true,
@@ -87,17 +99,17 @@ func TestMultilinePreviousNegate(t *testing.T) {
   codec.Event(4, 5, "ANOTHER line")
   codec.Event(6, 7, "DEBUG Next line")
 
-  if lines != 1 {
-    gt.Logf("Wrong line count received")
-    gt.FailNow()
+  if multiline_lines != 1 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
   }
 }
 
 func TestMultilinePreviousTimeout(t *testing.T) {
-  gt = t
-  lines = 0
+  multiline_t = t
+  multiline_lines = 0
 
-  codec := createCodec(map[string]interface{}{
+  codec := createMultilineCodec(map[string]interface{}{
     "pattern": "^(ANOTHER|NEXT) ",
     "what": "previous",
     "negate": false,
@@ -113,17 +125,17 @@ func TestMultilinePreviousTimeout(t *testing.T) {
   // Allow 7 seconds
   time.Sleep(7 * time.Second)
 
-  if lines != 2 {
-    gt.Logf("Wrong line count received")
-    gt.FailNow()
+  if multiline_lines != 2 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
   }
 }
 
 func TestMultilineNext(t *testing.T) {
-  gt = t
-  lines = 0
+  multiline_t = t
+  multiline_lines = 0
 
-  codec := createCodec(map[string]interface{}{
+  codec := createMultilineCodec(map[string]interface{}{
     "pattern": "^(DEBUG|NEXT) ",
     "what": "next",
     "negate": false,
@@ -135,17 +147,17 @@ func TestMultilineNext(t *testing.T) {
   codec.Event(4, 5, "ANOTHER line")
   codec.Event(6, 7, "DEBUG Next line")
 
-  if lines != 1 {
-    gt.Logf("Wrong line count received")
-    gt.FailNow()
+  if multiline_lines != 1 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
   }
 }
 
 func TestMultilineNextNegate(t *testing.T) {
-  gt = t
-  lines = 0
+  multiline_t = t
+  multiline_lines = 0
 
-  codec := createCodec(map[string]interface{}{
+  codec := createMultilineCodec(map[string]interface{}{
     "pattern": "^ANOTHER ",
     "what": "next",
     "negate": true,
@@ -157,34 +169,35 @@ func TestMultilineNextNegate(t *testing.T) {
   codec.Event(4, 5, "ANOTHER line")
   codec.Event(6, 7, "DEBUG Next line")
 
-  if lines != 1 {
-    gt.Logf("Wrong line count received")
-    gt.FailNow()
+  if multiline_lines != 1 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
   }
 }
 
 func checkMultilineMaxBytes(start_offset int64, end_offset int64, text string) {
-  lines++
+  multiline_lines++
 
-  if lines == 1 {
+  if multiline_lines == 1 {
     if text != "DEBUG First line\nsecond line\nthi" {
-      gt.Logf("Event data incorrect [% X]", text)
-      gt.FailNow()
+      multiline_t.Logf("Event data incorrect [% X]", text)
+      multiline_t.FailNow()
     }
+
     return
   }
 
   if text != "rd line" {
-    gt.Logf("Second event data incorrect [% X]", text)
-    gt.FailNow()
+    multiline_t.Logf("Second event data incorrect [% X]", text)
+    multiline_t.FailNow()
   }
 }
 
 func TestMultilineMaxBytes(t *testing.T) {
-  gt = t
-  lines = 0
+  multiline_t = t
+  multiline_lines = 0
 
-  codec := createCodec(map[string]interface{}{
+  codec := createMultilineCodec(map[string]interface{}{
     "max multiline bytes": int64(32),
     "pattern": "^DEBUG ",
     "negate": true,
@@ -196,8 +209,8 @@ func TestMultilineMaxBytes(t *testing.T) {
   codec.Event(4, 5, "third line")
   codec.Event(6, 7, "DEBUG Next line")
 
-  if lines != 2 {
-    gt.Logf("Wrong line count received")
-    gt.FailNow()
+  if multiline_lines != 2 {
+    t.Logf("Wrong line count received")
+    t.FailNow()
   }
 }

From 8aff403921524d32cba73a93fb56fabc6160b1a2 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 15:08:49 +0000
Subject: [PATCH 34/46] Drop multiline test from makefile. Locking for previous
 timeout test. Strip unused spec library

---
 Makefile                            |  2 +-
 spec/lib/logfile/multiline.rb       | 47 -----------------------------
 src/lc-lib/codecs/multiline_test.go | 18 +++++++++++
 3 files changed, 19 insertions(+), 48 deletions(-)
 delete mode 100644 spec/lib/logfile/multiline.rb

diff --git a/Makefile b/Makefile
index 8df171ca..7b0ac21e 100644
--- a/Makefile
+++ b/Makefile
@@ -7,7 +7,7 @@ export GOPATH := $(GOPATH)
 TAGS :=
 BINS := bin/log-courier bin/lc-tlscert bin/lc-admin
 GOTESTS := log-courier lc-tlscert lc-admin lc-lib/...
-TESTS := spec/courier_spec.rb spec/tcp_spec.rb spec/gem_spec.rb spec/multiline_spec.rb
+TESTS := spec/courier_spec.rb spec/tcp_spec.rb spec/gem_spec.rb
 
 ifneq (,$(findstring curvekey,$(MAKECMDGOALS)))
 with := zmq4
diff --git a/spec/lib/logfile/multiline.rb b/spec/lib/logfile/multiline.rb
deleted file mode 100644
index 9a6ff948..00000000
--- a/spec/lib/logfile/multiline.rb
+++ /dev/null
@@ -1,47 +0,0 @@
-# encoding: utf-8
-
-# Copyright 2014 Jason Woods.
-#
-# 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.
-
-class LogFile
-  # Multiline replaced the event checks so they match multiline events
-  class Multiline < LogFile
-    def log(num = 1)
-      num.times do |i|
-        i += @count + @next
-        # This pattern is chosen specifically as we can match it with all the different match types
-        @file.puts 'BEGIN ' + @path + " test event #{i}"
-        @file.puts " line 2 of test event #{i}"
-        @file.puts " line 3 of test event #{i}"
-        @file.puts ' END of test event'
-      end
-      @file.flush
-      @count += num
-      self
-    end
-
-    def skip_one
-      @count -= 1
-    end
-
-    def logged?(event: event, check_file: true, check_order: true)
-      return false if event['host'] != @host
-      return false if check_file && event['path'] != @path
-      return false if event['message'] != 'BEGIN ' + @path + " test event #{@next}" + $/ + " line 2 of test event #{@next}" + $/ + " line 3 of test event #{@next}" + $/ + ' END of test event'
-      @count -= 1
-      @next += 1
-      true
-    end
-  end
-end
diff --git a/src/lc-lib/codecs/multiline_test.go b/src/lc-lib/codecs/multiline_test.go
index 56b48b4b..7c8069a5 100644
--- a/src/lc-lib/codecs/multiline_test.go
+++ b/src/lc-lib/codecs/multiline_test.go
@@ -2,12 +2,14 @@ package codecs
 
 import (
   "lc-lib/core"
+  "sync"
   "testing"
   "time"
 )
 
 var multiline_t *testing.T
 var multiline_lines int
+var multiline_lock sync.Mutex
 
 func createMultilineCodec(unused map[string]interface{}, callback core.CodecCallbackFunc, t *testing.T) core.Codec {
   config := core.NewConfig()
@@ -24,6 +26,8 @@ func createMultilineCodec(unused map[string]interface{}, callback core.CodecCall
 }
 
 func checkMultiline(start_offset int64, end_offset int64, text string) {
+  multiline_lock.Lock()
+  defer multiline_lock.Unlock()
   multiline_lines++
 
   if multiline_lines == 1 {
@@ -122,13 +126,27 @@ func TestMultilinePreviousTimeout(t *testing.T) {
   codec.Event(4, 5, "ANOTHER line")
   codec.Event(6, 7, "DEBUG Next line")
 
+  // Allow 3 seconds
+  time.Sleep(3 * time.Second)
+
+  multiline_lock.Lock()
+  if multiline_lines != 1 {
+    t.Logf("Timeout triggered too early")
+    t.FailNow()
+  }
+  multiline_lock.Unlock()
+
   // Allow 7 seconds
   time.Sleep(7 * time.Second)
 
+  multiline_lock.Lock()
   if multiline_lines != 2 {
     t.Logf("Wrong line count received")
     t.FailNow()
   }
+  multiline_lock.Unlock()
+
+  codec.Teardown()
 }
 
 func TestMultilineNext(t *testing.T) {

From 4b31c5ea098014cc42fd6654117872e520a158ba Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 16:45:52 +0000
Subject: [PATCH 35/46] Fix ruby bundle install on jrprofile

---
 Makefile | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Makefile b/Makefile
index 7b0ac21e..3e09db90 100644
--- a/Makefile
+++ b/Makefile
@@ -73,7 +73,7 @@ vendor/bundle/.GemfileModT: Gemfile
 	bundle install --path vendor/bundle
 	@touch $@
 
-jrprofile: | all vendor/bundle/.GemfileModT
+jrprofile: | all vendor/bundle/.GemfileJRubyModT
 	jruby --profile -G vendor/bundle/jruby/1.9/bin/rspec spec/benchmark_spec.rb
 
 jrbenchmark: | all vendor/bundle/.GemfileJRubyModT

From e370ea1382c957e2722dad28708ea6789bb7d4b7 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 16:46:34 +0000
Subject: [PATCH 36/46] Fix a spooler race condition which could trigger spool
 timeout flush one or more times immediately after a max spool size flush
 containing only single figure event counts

---
 src/lc-lib/spooler/spooler.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/src/lc-lib/spooler/spooler.go b/src/lc-lib/spooler/spooler.go
index 20a1d816..709520d4 100644
--- a/src/lc-lib/spooler/spooler.go
+++ b/src/lc-lib/spooler/spooler.go
@@ -142,6 +142,13 @@ func (s *Spooler) sendSpool() bool {
 
 func (s *Spooler) resetTimer() {
   s.timer_start = time.Now()
+
+  // Stop the timer, and ensure the channel is empty before restarting it
+  s.timer.Stop()
+  select {
+  case <-s.timer.C:
+  default:
+  }
   s.timer.Reset(s.config.SpoolTimeout)
 }
 

From 5fc464d06941fa5ff9baec886fa8cc0c9a2fe72e Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 17:00:00 +0000
Subject: [PATCH 37/46] Fix MultiJson::ParseError exception capture

---
 lib/log-courier/server.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/log-courier/server.rb b/lib/log-courier/server.rb
index 08df68d8..41fc2b2b 100644
--- a/lib/log-courier/server.rb
+++ b/lib/log-courier/server.rb
@@ -158,7 +158,7 @@ def process_jdat(message, comm, event_queue)
         # Decode the JSON
         begin
           event = @json_adapter.load(data_buf, @json_options)
-        rescue MultiJson::ParserError => e
+        rescue MultiJson::ParseError => e
           @logger.warn("[LogCourierServer] JSON parse failure, falling back to plain-text: #{e}") unless @logger.nil?
           event = { 'message' => data_buf }
         end

From e564810ddbe090814f0ae5db5229d988434e4319 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 17:00:17 +0000
Subject: [PATCH 38/46] Implement ZMQ connection close monitor event

---
 src/lc-lib/transports/zmq.go | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/src/lc-lib/transports/zmq.go b/src/lc-lib/transports/zmq.go
index 9e59f58c..cffbc35d 100644
--- a/src/lc-lib/transports/zmq.go
+++ b/src/lc-lib/transports/zmq.go
@@ -106,6 +106,12 @@ func (e *ZMQEvent) Log() {
     } else {
       log.Info("Attempting to connect to %s", e.data)
     }
+  case zmq.EVENT_CLOSED:
+    if e.data == "" {
+      log.Error("Connection closed")
+    } else {
+      log.Error("Connection to %s closed", e.data)
+    }
   case zmq.EVENT_DISCONNECTED:
     if e.data == "" {
       log.Error("Lost connection")

From 320a3c1904e4b5a6366f7ac19ea82aa440667bbd Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 19:28:06 +0000
Subject: [PATCH 39/46] Tweak debug message for multipart problem

---
 lib/log-courier/server_zmq.rb | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index 37287a14..73b60657 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -132,7 +132,7 @@ def run(&block)
           @return_route = []
           data.delete_if do |msg|
             reset_timeout
-            break if msg == ""
+            break if msg == ''
             @return_route.push msg
             true
           end
@@ -154,7 +154,8 @@ def run(&block)
               i = 0
               data.each do |msg|
                 i += 1
-                @logger.debug "[LogCourierServer] Part #{i}: #{msg[0..31].gsub(/[^[:print:]]/,'.')}"
+                part = msg[0..31].gsub(/[^[:print:]]/, '.')
+                @logger.debug "[LogCourierServer] Part #{i}: #{part.length}:[#{part}]"
               end
             end
           else

From 872465bdab140c5cb88c52f2e734a588d9fa831b Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 19:28:29 +0000
Subject: [PATCH 40/46] Improve shutdown times on JRuby with a funky hack :\

---
 lib/log-courier/server_zmq.rb | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index 73b60657..38f30171 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -119,7 +119,8 @@ def run(&block)
               @poller.deregister @socket, ZMQ::POLLIN | ZMQ::POLLOUT
               @poller.register @socket, ZMQ::POLLIN
               while @poller.poll(1_000) == 0
-                raise TimeoutError if Time.now.to_i >= @timeout
+                # Using this inner while triggers pollThreadEvents in JRuby which checks for Thread.raise immediately
+                raise TimeoutError while Time.now.to_i >= @timeout
               end
               next
             end
@@ -229,7 +230,8 @@ def send_with_poll(data, more = false)
         @poller.deregister @socket, ZMQ::POLLIN | ZMQ::POLLOUT
         @poller.register @socket, ZMQ::POLLOUT
         while @poller.poll(1_000) == 0
-          raise TimeoutError if Time.now.to_i >= @timeout
+          # Using this inner while triggers pollThreadEvents in JRuby which checks for Thread.raise immediately
+          raise TimeoutError while Time.now.to_i >= @timeout
         end
       end
     end

From 9060beabd5df34f690d865d37a155090d60b9433 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 21:44:37 +0000
Subject: [PATCH 41/46] Prevent memory issues during stress tests with a sized
 queue instead of unbounded queue

---
 spec/courier_spec.rb       | 1 +
 spec/lib/helpers/common.rb | 2 +-
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/spec/courier_spec.rb b/spec/courier_spec.rb
index f93aba7b..5bf23576 100644
--- a/spec/courier_spec.rb
+++ b/spec/courier_spec.rb
@@ -37,6 +37,7 @@
     }
     config
 
+    # Remember the sized queue we use for test buffering is only 10_000 lines
     5_000.times do |i|
       @log_courier.puts "stdin line test #{i}"
     end
diff --git a/spec/lib/helpers/common.rb b/spec/lib/helpers/common.rb
index f39f62c9..3f7342d0 100644
--- a/spec/lib/helpers/common.rb
+++ b/spec/lib/helpers/common.rb
@@ -44,7 +44,7 @@
     # When we add a file we log it here, so after we can remove them
     @files = []
 
-    @event_queue = Queue.new
+    @event_queue = SizedQueue.new 10_000
 
     @servers = {}
     @server_counts = {}

From 3eaf3d7f60e5dcaa7eaac9deb02b63fb97c4b24d Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 21:45:35 +0000
Subject: [PATCH 42/46] Log JRuby native exceptions and correctly reraise

---
 lib/log-courier/client.rb     | 2 ++
 lib/log-courier/client_tls.rb | 3 ++-
 lib/log-courier/server.rb     | 2 ++
 lib/log-courier/server_tcp.rb | 5 +++++
 lib/log-courier/server_zmq.rb | 3 ++-
 5 files changed, 13 insertions(+), 2 deletions(-)

diff --git a/lib/log-courier/client.rb b/lib/log-courier/client.rb
index 409af0cc..887f51d9 100644
--- a/lib/log-courier/client.rb
+++ b/lib/log-courier/client.rb
@@ -22,6 +22,8 @@
 require 'thread'
 require 'zlib'
 
+class NativeException; end
+
 module LogCourier
   # TODO: Make these shared
   class ClientShutdownSignal < StandardError; end
diff --git a/lib/log-courier/client_tls.rb b/lib/log-courier/client_tls.rb
index bb267223..141d3a08 100644
--- a/lib/log-courier/client_tls.rb
+++ b/lib/log-courier/client_tls.rb
@@ -209,9 +209,10 @@ def tls_connect
     rescue ClientShutdownSignal
       # Just shutdown
       0
-    rescue => e
+    rescue StandardError, NativeException => e
       @logger.warn("[LogCourierClient] Unknown connection failure to #{@options[:addresses][0]}:#{@options[:port]}: #{e}") unless @logger.nil?
       @logger.warn("[LogCourierClient] #{e.backtrace}: #{e.message} (#{e.class})") unless @logger.nil?
+      raise e
     end
   end
 end
diff --git a/lib/log-courier/server.rb b/lib/log-courier/server.rb
index 41fc2b2b..9c2d2036 100644
--- a/lib/log-courier/server.rb
+++ b/lib/log-courier/server.rb
@@ -22,6 +22,8 @@
 require 'thread'
 require 'zlib'
 
+class NativeException; end
+
 module LogCourier
   class TimeoutError < StandardError; end
   class ShutdownSignal < StandardError; end
diff --git a/lib/log-courier/server_tcp.rb b/lib/log-courier/server_tcp.rb
index caf5423f..ec6f5340 100644
--- a/lib/log-courier/server_tcp.rb
+++ b/lib/log-courier/server_tcp.rb
@@ -139,6 +139,11 @@ def run(&block)
     rescue ShutdownSignal
       # Capture shutting down signal
       0
+    rescue StandardError, NativeException => e
+      # Some other unknown problem
+      @logger.warn("[LogCourierServer] Unknown error: #{e}") unless @logger.nil?
+      @logger.warn("[LogCourierServer] #{e.backtrace}: #{e.message} (#{e.class})") unless @logger.nil?
+      raise e
     ensure
       # Raise shutdown in all client threads and join then
       client_threads.each do |_, thr|
diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index 38f30171..e98ee9c0 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -173,10 +173,11 @@ def run(&block)
     rescue ShutdownSignal
       # Shutting down
       @logger.warn('[LogCourierServer] Server shutting down') unless @logger.nil?
-    rescue => e
+    rescue StandardError, NativeException => e
       # Some other unknown problem
       @logger.warn("[LogCourierServer] Unknown error: #{e}") unless @logger.nil?
       @logger.warn("[LogCourierServer] #{e.backtrace}: #{e.message} (#{e.class})") unless @logger.nil?
+      raise e
     ensure
       @socket.close
       @context.terminate

From 195e0760f4ef7f2d93062628f94f98017fe93625 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 21:47:09 +0000
Subject: [PATCH 43/46] Normalise LogCourier gem exception names between client
 and server

---
 lib/log-courier/client.rb     | 19 +++++++++----------
 lib/log-courier/client_tls.rb | 12 ++++++------
 2 files changed, 15 insertions(+), 16 deletions(-)

diff --git a/lib/log-courier/client.rb b/lib/log-courier/client.rb
index 887f51d9..ec546461 100644
--- a/lib/log-courier/client.rb
+++ b/lib/log-courier/client.rb
@@ -25,9 +25,8 @@
 class NativeException; end
 
 module LogCourier
-  # TODO: Make these shared
-  class ClientShutdownSignal < StandardError; end
-  class ClientProtocolError < StandardError; end
+  class ShutdownSignal < StandardError; end
+  class ProtocolError < StandardError; end
 
   # Describes a pending payload
   class PendingPayload
@@ -95,8 +94,8 @@ def publish(event)
 
     def shutdown
       # Raise a shutdown signal in the spooler and wait for it
-      @spooler_thread.raise ClientShutdownSignal
-      @io_thread.raise ClientShutdownSignal
+      @spooler_thread.raise ShutdownSignal
+      @io_thread.raise ShutdownSignal
       @spooler_thread.join
       @io_thread.join
     end
@@ -125,7 +124,7 @@ def run_spooler
           @io_control << ['E', spooled]
         end
       end
-    rescue ClientShutdownSignal
+    rescue ShutdownSignal
       # Just shutdown
       0
     end
@@ -226,13 +225,13 @@ def run_io
             # Reset keepalive timeout
             reset_keepalive
           end
-        rescue ClientProtocolError => e
+        rescue ProtocolError => e
           # Reconnect required due to a protocol error
           @logger.warn("[LogCourierClient] Protocol error: #{e}") unless @logger.nil?
         rescue TimeoutError
           # Reconnect due to timeout
           @logger.warn('[LogCourierClient] Timeout occurred') unless @logger.nil?
-        rescue ClientShutdownSignal
+        rescue ShutdownSignal
           # Shutdown, break out
           break
         rescue => e
@@ -313,7 +312,7 @@ def buffer_jdat_data_event(buffer, event)
     def process_pong(message)
       # Sanity
       if message.length != 0
-        raise ClientProtocolError, "Unexpected data attached to pong message (#{message.length})"
+        raise ProtocolError, "Unexpected data attached to pong message (#{message.length})"
       end
 
       # No longer pending a PONG
@@ -323,7 +322,7 @@ def process_pong(message)
     def process_ackn(message)
       # Sanity
       if message.length != 20
-        raise ClientProtocolError, "ACKN message size invalid (#{message.length})"
+        raise ProtocolError, "ACKN message size invalid (#{message.length})"
       end
 
       # Grab nonce
diff --git a/lib/log-courier/client_tls.rb b/lib/log-courier/client_tls.rb
index 141d3a08..fdf0bd91 100644
--- a/lib/log-courier/client_tls.rb
+++ b/lib/log-courier/client_tls.rb
@@ -54,7 +54,7 @@ def initialize(options = {})
     def connect(io_control)
       begin
         tls_connect
-      rescue ClientShutdownSignal
+      rescue ShutdownSignal
         raise
       rescue
         # TODO: Make this configurable
@@ -74,9 +74,9 @@ def connect(io_control)
     end
 
     def disconnect
-      @send_thread.raise ClientShutdownSignal
+      @send_thread.raise ShutdownSignal
       @send_thread.join
-      @recv_thread.raise ClientShutdownSignal
+      @recv_thread.raise ShutdownSignal
       @recv_thread.join
     end
 
@@ -111,7 +111,7 @@ def run_send(io_control)
     rescue OpenSSL::SSL::SSLError, IOError, Errno::ECONNRESET => e
       @logger.warn("[LogCourierClient] SSL write error: #{e}") unless @logger.nil?
       io_control << ['F']
-    rescue ClientShutdownSignal
+    rescue ShutdownSignal
       # Just shutdown
     rescue => e
       @logger.warn("[LogCourierClient] Unknown SSL write error: #{e}") unless @logger.nil?
@@ -147,7 +147,7 @@ def run_recv(io_control)
     rescue EOFError
       @logger.warn("[LogCourierClient] Connection closed by server") unless @logger.nil?
       io_control << ['F']
-    rescue ClientShutdownSignal
+    rescue ShutdownSignal
       # Just shutdown
     rescue => e
       @logger.warn("[LogCourierClient] Unknown SSL read error: #{e}") unless @logger.nil?
@@ -206,7 +206,7 @@ def tls_connect
       socket
     rescue OpenSSL::SSL::SSLError, IOError, Errno::ECONNRESET => e
       @logger.warn("[LogCourierClient] Connection to #{@options[:addresses][0]}:#{@options[:port]} failed: #{e}") unless @logger.nil?
-    rescue ClientShutdownSignal
+    rescue ShutdownSignal
       # Just shutdown
       0
     rescue StandardError, NativeException => e

From 3f1185013a5d29b68c852349362dfbd780ccbab5 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 22:35:02 +0000
Subject: [PATCH 44/46] Fixes #63 - delete_if issue in JRuby 1.17.11 (not
 present in 1.17.16) - switched to more compact method with shift
 https://github.com/jruby/jruby/commit/b653e37a67be8b94be87c21b0af52ebf055cf014

---
 lib/log-courier/server_zmq.rb | 14 +++++---------
 1 file changed, 5 insertions(+), 9 deletions(-)

diff --git a/lib/log-courier/server_zmq.rb b/lib/log-courier/server_zmq.rb
index e98ee9c0..36b8569f 100644
--- a/lib/log-courier/server_zmq.rb
+++ b/lib/log-courier/server_zmq.rb
@@ -110,6 +110,7 @@ def run(&block)
         begin
           begin
             # Try to receive a message
+            reset_timeout
             data = []
             rc = @socket.recv_strings(data, ZMQ::DONTWAIT)
             unless ZMQ::Util.resultcode_ok?(rc)
@@ -131,12 +132,7 @@ def run(&block)
 
           # Save the routing information that appears before the null messages
           @return_route = []
-          data.delete_if do |msg|
-            reset_timeout
-            break if msg == ''
-            @return_route.push msg
-            true
-          end
+          @return_route.push data.shift until data.length == 0 || data[0] == ''
 
           if data.length == 0
             @logger.warn '[LogCourierServer] Invalid message: no data' unless @logger.nil?
@@ -206,14 +202,14 @@ def recv(data)
     end
 
     def send(signature, message)
-      reset_timeout
       data = signature + [message.length].pack('N') + message
 
       # Send the return route and then the message
+      reset_timeout
       @return_route.each do |msg|
         send_with_poll msg, true
       end
-      send_with_poll "", true
+      send_with_poll '', true
       send_with_poll data
     end
 
@@ -237,7 +233,7 @@ def send_with_poll(data, more = false)
       end
     end
 
-    def reset_timeout()
+    def reset_timeout
       # TODO: Make configurable?
       @timeout = Time.now.to_i + 1_800
     end

From 52ed5ce01564ffa0c20fbfa981636b75602ab1bf Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 22:35:14 +0000
Subject: [PATCH 45/46] Changelog update

---
 docs/ChangeLog.md | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/docs/ChangeLog.md b/docs/ChangeLog.md
index 66f9b65b..c69ef7a5 100644
--- a/docs/ChangeLog.md
+++ b/docs/ChangeLog.md
@@ -24,10 +24,17 @@
 * Implement gems for the new Logstash plugin system (#60)
 * Fix gem build failing on develop branch with old rubygems versions due to a
 malformed version string (#62)
+* Fix ZeroMQ transports in the ruby gem with Logstash 1.4.x (#63)
+* Fix build issue with ZeroMQ 3.2 and `make with=zmq3`
+* Fix partial acknowledgements not being passed to registrar and persisted to
+disk
+* Fix a race condition when the spooler flushes to prevent a timeout occurring
+one or more times after a flush due to size
 * Print informational messages containing ZMQ library version information during
 gem and log-courier startup to aid in diagnostics
 * Raise a friendly error when trying to use the zmq transport in the Log Courier
 gem with incompatible versions of libzmq
+* Various fixes and improvements to log-courier, gem, build and tests
 
 ## 1.0
 

From da940769a808ee04d295ae8f436c552bfb404f07 Mon Sep 17 00:00:00 2001
From: Driskell <devel@jasonwoods.me.uk>
Date: Thu, 30 Oct 2014 22:56:05 +0000
Subject: [PATCH 46/46] Release v1.1

---
 docs/ChangeLog.md                   | 6 +++---
 log-courier.gemspec                 | 2 +-
 logstash-input-log-courier.gemspec  | 4 ++--
 logstash-output-log-courier.gemspec | 4 ++--
 src/lc-lib/core/version.go          | 2 +-
 5 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/docs/ChangeLog.md b/docs/ChangeLog.md
index c69ef7a5..9c8b737e 100644
--- a/docs/ChangeLog.md
+++ b/docs/ChangeLog.md
@@ -4,7 +4,7 @@
 <!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
 **Table of Contents**  *generated with [DocToc](http://doctoc.herokuapp.com/)*
 
-- [Latest Development](#latest-development)
+- [1.1](#11)
 - [1.0](#10)
 - [0.15](#015)
 - [0.14](#014)
@@ -17,9 +17,9 @@
 
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
 
-## Latest Development
+## 1.1
 
-*TBC*
+*30th October 2014*
 
 * Implement gems for the new Logstash plugin system (#60)
 * Fix gem build failing on develop branch with old rubygems versions due to a
diff --git a/log-courier.gemspec b/log-courier.gemspec
index 2218a076..339d6b91 100644
--- a/log-courier.gemspec
+++ b/log-courier.gemspec
@@ -1,6 +1,6 @@
 Gem::Specification.new do |gem|
   gem.name              = 'log-courier'
-  gem.version           = '1.0'
+  gem.version           = '1.1'
   gem.description       = 'Log Courier library'
   gem.summary           = 'Receive events from Log Courier and transmit between LogStash instances'
   gem.homepage          = 'https://github.com/driskell/log-courier'
diff --git a/logstash-input-log-courier.gemspec b/logstash-input-log-courier.gemspec
index 185a4426..3fe867ed 100644
--- a/logstash-input-log-courier.gemspec
+++ b/logstash-input-log-courier.gemspec
@@ -1,6 +1,6 @@
 Gem::Specification.new do |gem|
   gem.name              = 'logstash-input-log-courier'
-  gem.version           = '1.0'
+  gem.version           = '1.1'
   gem.description       = 'Log Courier Input Logstash Plugin'
   gem.summary           = 'Receive events from Log Courier and Logstash using the Log Courier protocol'
   gem.homepage          = 'https://github.com/driskell/log-courier'
@@ -16,5 +16,5 @@ Gem::Specification.new do |gem|
   gem.metadata = { 'logstash_plugin' => 'true', 'group' => 'input' }
 
   gem.add_runtime_dependency 'logstash', '~> 1.4'
-  gem.add_runtime_dependency 'log-courier', '= 1.0'
+  gem.add_runtime_dependency 'log-courier', '= 1.1'
 end
diff --git a/logstash-output-log-courier.gemspec b/logstash-output-log-courier.gemspec
index b9ec8075..194f48d0 100644
--- a/logstash-output-log-courier.gemspec
+++ b/logstash-output-log-courier.gemspec
@@ -1,6 +1,6 @@
 Gem::Specification.new do |gem|
   gem.name              = 'logstash-output-log-courier'
-  gem.version           = '1.0'
+  gem.version           = '1.1'
   gem.description       = 'Log Courier Output Logstash Plugin'
   gem.summary           = 'Transmit events from one Logstash instance to another using the Log Courier protocol'
   gem.homepage          = 'https://github.com/driskell/log-courier'
@@ -16,5 +16,5 @@ Gem::Specification.new do |gem|
   gem.metadata = { 'logstash_plugin' => 'true', 'group' => 'input' }
 
   gem.add_runtime_dependency 'logstash', '~> 1.4'
-  gem.add_runtime_dependency 'log-courier', '= 1.0'
+  gem.add_runtime_dependency 'log-courier', '= 1.1'
 end
diff --git a/src/lc-lib/core/version.go b/src/lc-lib/core/version.go
index 00d92f59..00d966cb 100644
--- a/src/lc-lib/core/version.go
+++ b/src/lc-lib/core/version.go
@@ -16,4 +16,4 @@
 
 package core
 
-const Log_Courier_Version string = "v1.0"
+const Log_Courier_Version string = "v1.1"