Merge pull request #93 from tendermint/release/0.5.0

Release/0.5.0

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 0.5.0 (December 5, 2017)
+
+BREAKING:
+ - [common] replace Service#Start, Service#Stop first return value (bool) with an
+   error (ErrAlreadyStarted, ErrAlreadyStopped)
+ - [common] replace Service#Reset first return value (bool) with an error
+ - [process] removed
+
+FEATURES:
+ - [common] IntInSlice and StringInSlice functions
+ - [pubsub/query] introduce `Condition` struct, expose `Operator`, and add `query.Conditions()`
+
 ## 0.4.1 (November 27, 2017)
 
 FEATURES:

LICENSE

@@ -0,0 +1,193 @@
+Tendermint Libraries
+Copyright (C) 2017 Tendermint
+
+
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        https://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   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
+
+       https://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.

Makefile

@@ -4,14 +4,16 @@ GOTOOLS = \
 	github.com/Masterminds/glide \
 	github.com/alecthomas/gometalinter
 
+PACKAGES=$(shell go list ./... | grep -v '/vendor/')
 REPO:=github.com/tendermint/tmlibs
 
 all: test
 
-NOVENDOR = go list github.com/tendermint/tmlibs/... | grep -v /vendor/
-
 test:
-	go test `glide novendor`
+	@echo "--> Running linter"
+	@make metalinter_test
+	@echo "--> Running go test"
+	@go test $(PACKAGES)
 
 get_vendor_deps: ensure_tools
 	@rm -rf vendor/
@@ -20,16 +22,14 @@ get_vendor_deps: ensure_tools
 
 ensure_tools:
 	go get $(GOTOOLS)
-
-metalinter: ensure_tools
 	@gometalinter --install
+
+metalinter: 
 	gometalinter --vendor --deadline=600s --enable-all --disable=lll ./...
 
-metalinter_test: ensure_tools
-	@gometalinter --install
+metalinter_test: 
 	gometalinter --vendor --deadline=600s --disable-all  \
 		--enable=deadcode \
-		--enable=gas \
 		--enable=goconst \
 		--enable=gosimple \
 	 	--enable=ineffassign \
@@ -46,6 +46,7 @@ metalinter_test: ensure_tools
 		--enable=vet \
 		./...
 
+		#--enable=gas \
 		#--enable=aligncheck \
 		#--enable=dupl \
 		#--enable=errcheck \

circle.yml

@@ -15,7 +15,7 @@ dependencies:
 
 test:
   override:
-    - cd $PROJECT_PATH && make get_vendor_deps && make metalinter_test && bash ./test.sh
+    - cd $PROJECT_PATH && make get_vendor_deps && bash ./test.sh
   post:
     - cd "$PROJECT_PATH" && bash <(curl -s https://codecov.io/bash) -f coverage.txt
     - cd "$PROJECT_PATH" && mv coverage.txt "${CIRCLE_ARTIFACTS}"

common/int.go

@@ -53,3 +53,13 @@ func PutInt64BE(dest []byte, i int64) {
 func GetInt64BE(src []byte) int64 {
 	return int64(binary.BigEndian.Uint64(src))
 }
+
+// IntInSlice returns true if a is found in the list.
+func IntInSlice(a int, list []int) bool {
+	for _, b := range list {
+		if b == a {
+			return true
+		}
+	}
+	return false
+}

common/int_test.go

@@ -0,0 +1,14 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIntInSlice(t *testing.T) {
+	assert.True(t, IntInSlice(1, []int{1, 2, 3}))
+	assert.False(t, IntInSlice(4, []int{1, 2, 3}))
+	assert.True(t, IntInSlice(0, []int{0}))
+	assert.False(t, IntInSlice(0, []int{}))
+}

common/repeat_timer_test.go

@@ -0,0 +1,78 @@
+package common
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	// make govet noshadow happy...
+	asrt "github.com/stretchr/testify/assert"
+)
+
+type rCounter struct {
+	input chan time.Time
+	mtx   sync.Mutex
+	count int
+}
+
+func (c *rCounter) Increment() {
+	c.mtx.Lock()
+	c.count++
+	c.mtx.Unlock()
+}
+
+func (c *rCounter) Count() int {
+	c.mtx.Lock()
+	val := c.count
+	c.mtx.Unlock()
+	return val
+}
+
+// Read should run in a go-routine and
+// updates count by one every time a packet comes in
+func (c *rCounter) Read() {
+	for range c.input {
+		c.Increment()
+	}
+}
+
+func TestRepeat(test *testing.T) {
+	assert := asrt.New(test)
+
+	dur := time.Duration(50) * time.Millisecond
+	short := time.Duration(20) * time.Millisecond
+	// delay waits for cnt durations, an a little extra
+	delay := func(cnt int) time.Duration {
+		return time.Duration(cnt)*dur + time.Millisecond
+	}
+	t := NewRepeatTimer("bar", dur)
+
+	// start at 0
+	c := &rCounter{input: t.Ch}
+	go c.Read()
+	assert.Equal(0, c.Count())
+
+	// wait for 4 periods
+	time.Sleep(delay(4))
+	assert.Equal(4, c.Count())
+
+	// keep reseting leads to no firing
+	for i := 0; i < 20; i++ {
+		time.Sleep(short)
+		t.Reset()
+	}
+	assert.Equal(4, c.Count())
+
+	// after this, it still works normal
+	time.Sleep(delay(2))
+	assert.Equal(6, c.Count())
+
+	// after a stop, nothing more is sent
+	stopped := t.Stop()
+	assert.True(stopped)
+	time.Sleep(delay(7))
+	assert.Equal(6, c.Count())
+
+	// close channel to stop counter
+	close(t.Ch)
+}

common/service.go

@@ -1,23 +1,41 @@
 package common
 
 import (
+	"errors"
+	"fmt"
 	"sync/atomic"
 
 	"github.com/tendermint/tmlibs/log"
 )
 
+var (
+	ErrAlreadyStarted = errors.New("already started")
+	ErrAlreadyStopped = errors.New("already stopped")
+)
+
+// Service defines a service that can be started, stopped, and reset.
 type Service interface {
-	Start() (bool, error)
+	// Start the service.
+	// If it's already started or stopped, will return an error.
+	// If OnStart() returns an error, it's returned by Start()
+	Start() error
 	OnStart() error
 
-	Stop() bool
+	// Stop the service.
+	// If it's already stopped, will return an error.
+	// OnStop must never error.
+	Stop() error
 	OnStop()
 
-	Reset() (bool, error)
+	// Reset the service.
+	// Panics by default - must be overwritten to enable reset.
+	Reset() error
 	OnReset() error
 
+	// Return true if the service is running
 	IsRunning() bool
 
+	// String representation of the service
 	String() string
 
 	SetLogger(log.Logger)
@@ -94,11 +112,11 @@ func (bs *BaseService) SetLogger(l log.Logger) {
 }
 
 // Implements Servce
-func (bs *BaseService) Start() (bool, error) {
+func (bs *BaseService) Start() error {
 	if atomic.CompareAndSwapUint32(&bs.started, 0, 1) {
 		if atomic.LoadUint32(&bs.stopped) == 1 {
 			bs.Logger.Error(Fmt("Not starting %v -- already stopped", bs.name), "impl", bs.impl)
-			return false, nil
+			return ErrAlreadyStopped
 		} else {
 			bs.Logger.Info(Fmt("Starting %v", bs.name), "impl", bs.impl)
 		}
@@ -106,12 +124,12 @@ func (bs *BaseService) Start() (bool, error) {
 		if err != nil {
 			// revert flag
 			atomic.StoreUint32(&bs.started, 0)
-			return false, err
+			return err
 		}
-		return true, err
+		return nil
 	} else {
 		bs.Logger.Debug(Fmt("Not starting %v -- already started", bs.name), "impl", bs.impl)
-		return false, nil
+		return ErrAlreadyStarted
 	}
 }
 
@@ -121,15 +139,15 @@ func (bs *BaseService) Start() (bool, error) {
 func (bs *BaseService) OnStart() error { return nil }
 
 // Implements Service
-func (bs *BaseService) Stop() bool {
+func (bs *BaseService) Stop() error {
 	if atomic.CompareAndSwapUint32(&bs.stopped, 0, 1) {
 		bs.Logger.Info(Fmt("Stopping %v", bs.name), "impl", bs.impl)
 		bs.impl.OnStop()
 		close(bs.Quit)
-		return true
+		return nil
 	} else {
 		bs.Logger.Debug(Fmt("Stopping %v (ignoring: already stopped)", bs.name), "impl", bs.impl)
-		return false
+		return ErrAlreadyStopped
 	}
 }
 
@@ -139,17 +157,17 @@ func (bs *BaseService) Stop() bool {
 func (bs *BaseService) OnStop() {}
 
 // Implements Service
-func (bs *BaseService) Reset() (bool, error) {
+func (bs *BaseService) Reset() error {
 	if !atomic.CompareAndSwapUint32(&bs.stopped, 1, 0) {
 		bs.Logger.Debug(Fmt("Can't reset %v. Not stopped", bs.name), "impl", bs.impl)
-		return false, nil
+		return fmt.Errorf("can't reset running %s", bs.name)
 	}
 
 	// whether or not we've started, we can reset
 	atomic.CompareAndSwapUint32(&bs.started, 1, 0)
 
 	bs.Quit = make(chan struct{})
-	return true, bs.impl.OnReset()
+	return bs.impl.OnReset()
 }
 
 // Implements Service

common/service_test.go

@@ -2,23 +2,53 @@ package common
 
 import (
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
 )
 
-func TestBaseServiceWait(t *testing.T) {
+type testService struct {
+	BaseService
+}
 
-	type TestService struct {
-		BaseService
-	}
-	ts := &TestService{}
+func (testService) OnReset() error {
+	return nil
+}
+
+func TestBaseServiceWait(t *testing.T) {
+	ts := &testService{}
 	ts.BaseService = *NewBaseService(nil, "TestService", ts)
 	ts.Start()
 
+	waitFinished := make(chan struct{})
 	go func() {
-		ts.Stop()
+		ts.Wait()
+		waitFinished <- struct{}{}
 	}()
 
-	for i := 0; i < 10; i++ {
-		ts.Wait()
+	go ts.Stop()
+
+	select {
+	case <-waitFinished:
+		// all good
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("expected Wait() to finish within 100 ms.")
 	}
+}
+
+func TestBaseServiceReset(t *testing.T) {
+	ts := &testService{}
+	ts.BaseService = *NewBaseService(nil, "TestService", ts)
+	ts.Start()
+
+	err := ts.Reset()
+	require.Error(t, err, "expected cant reset service error")
+
+	ts.Stop()
+
+	err = ts.Reset()
+	require.NoError(t, err)
 
+	err = ts.Start()
+	require.NoError(t, err)
 }

common/string.go

@@ -43,3 +43,13 @@ func StripHex(s string) string {
 	}
 	return s
 }
+
+// StringInSlice returns true if a is found the list.
+func StringInSlice(a string, list []string) bool {
+	for _, b := range list {
+		if b == a {
+			return true
+		}
+	}
+	return false
+}

common/string_test.go

@@ -0,0 +1,14 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStringInSlice(t *testing.T) {
+	assert.True(t, StringInSlice("a", []string{"a", "b", "c"}))
+	assert.False(t, StringInSlice("d", []string{"a", "b", "c"}))
+	assert.True(t, StringInSlice("", []string{""}))
+	assert.False(t, StringInSlice("", []string{}))
+}

common/throttle_timer_test.go

@@ -0,0 +1,78 @@
+package common
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	// make govet noshadow happy...
+	asrt "github.com/stretchr/testify/assert"
+)
+
+type thCounter struct {
+	input chan struct{}
+	mtx   sync.Mutex
+	count int
+}
+
+func (c *thCounter) Increment() {
+	c.mtx.Lock()
+	c.count++
+	c.mtx.Unlock()
+}
+
+func (c *thCounter) Count() int {
+	c.mtx.Lock()
+	val := c.count
+	c.mtx.Unlock()
+	return val
+}
+
+// Read should run in a go-routine and
+// updates count by one every time a packet comes in
+func (c *thCounter) Read() {
+	for range c.input {
+		c.Increment()
+	}
+}
+
+func TestThrottle(test *testing.T) {
+	assert := asrt.New(test)
+
+	ms := 50
+	delay := time.Duration(ms) * time.Millisecond
+	longwait := time.Duration(2) * delay
+	t := NewThrottleTimer("foo", delay)
+
+	// start at 0
+	c := &thCounter{input: t.Ch}
+	assert.Equal(0, c.Count())
+	go c.Read()
+
+	// waiting does nothing
+	time.Sleep(longwait)
+	assert.Equal(0, c.Count())
+
+	// send one event adds one
+	t.Set()
+	time.Sleep(longwait)
+	assert.Equal(1, c.Count())
+
+	// send a burst adds one
+	for i := 0; i < 5; i++ {
+		t.Set()
+	}
+	time.Sleep(longwait)
+	assert.Equal(2, c.Count())
+
+	// send 12, over 2 delay sections, adds 3
+	short := time.Duration(ms/5) * time.Millisecond
+	for i := 0; i < 13; i++ {
+		t.Set()
+		time.Sleep(short)
+	}
+	time.Sleep(longwait)
+	assert.Equal(5, c.Count())
+
+	close(t.Ch)
+}

events/events_test.go

@@ -13,8 +13,8 @@ import (
 // listener to an event, and sends a string "data".
 func TestAddListenerForEventFireOnce(t *testing.T) {
 	evsw := NewEventSwitch()
-	started, err := evsw.Start()
-	if !started || err != nil {
+	err := evsw.Start()
+	if err != nil {
 		t.Errorf("Failed to start EventSwitch, error: %v", err)
 	}
 	messages := make(chan EventData)
@@ -33,8 +33,8 @@ func TestAddListenerForEventFireOnce(t *testing.T) {
 // listener to an event, and sends a thousand integers.
 func TestAddListenerForEventFireMany(t *testing.T) {
 	evsw := NewEventSwitch()
-	started, err := evsw.Start()
-	if !started || err != nil {
+	err := evsw.Start()
+	if err != nil {
 		t.Errorf("Failed to start EventSwitch, error: %v", err)
 	}
 	doneSum := make(chan uint64)
@@ -62,8 +62,8 @@ func TestAddListenerForEventFireMany(t *testing.T) {
 // of the three events.
 func TestAddListenerForDifferentEvents(t *testing.T) {
 	evsw := NewEventSwitch()
-	started, err := evsw.Start()
-	if !started || err != nil {
+	err := evsw.Start()
+	if err != nil {
 		t.Errorf("Failed to start EventSwitch, error: %v", err)
 	}
 	doneSum := make(chan uint64)
@@ -107,8 +107,8 @@ func TestAddListenerForDifferentEvents(t *testing.T) {
 // for each of the three events.
 func TestAddDifferentListenerForDifferentEvents(t *testing.T) {
 	evsw := NewEventSwitch()
-	started, err := evsw.Start()
-	if !started || err != nil {
+	err := evsw.Start()
+	if err != nil {
 		t.Errorf("Failed to start EventSwitch, error: %v", err)
 	}
 	doneSum1 := make(chan uint64)
@@ -167,8 +167,8 @@ func TestAddDifferentListenerForDifferentEvents(t *testing.T) {
 // the listener and fires a thousand integers for the second event.
 func TestAddAndRemoveListener(t *testing.T) {
 	evsw := NewEventSwitch()
-	started, err := evsw.Start()
-	if !started || err != nil {
+	err := evsw.Start()
+	if err != nil {
 		t.Errorf("Failed to start EventSwitch, error: %v", err)
 	}
 	doneSum1 := make(chan uint64)
@@ -212,8 +212,8 @@ func TestAddAndRemoveListener(t *testing.T) {
 // TestRemoveListener does basic tests on adding and removing
 func TestRemoveListener(t *testing.T) {
 	evsw := NewEventSwitch()
-	started, err := evsw.Start()
-	if !started || err != nil {
+	err := evsw.Start()
+	if err != nil {
 		t.Errorf("Failed to start EventSwitch, error: %v", err)
 	}
 	count := 10
@@ -265,8 +265,8 @@ func TestRemoveListener(t *testing.T) {
 // `go test -race`, to examine for possible race conditions.
 func TestRemoveListenersAsync(t *testing.T) {
 	evsw := NewEventSwitch()
-	started, err := evsw.Start()
-	if !started || err != nil {
+	err := evsw.Start()
+	if err != nil {
 		t.Errorf("Failed to start EventSwitch, error: %v", err)
 	}
 	doneSum1 := make(chan uint64)

glide.lock

@@ -1,10 +1,10 @@
-hash: 6efda1f3891a7211fc3dc1499c0079267868ced9739b781928af8e225420f867
-updated: 2017-08-11T20:28:34.550901198Z
+hash: 1f3d3426e823e4a8e6d4473615fcc86c767bbea6da9114ea1e7e0a9f0ccfa129
+updated: 2017-12-05T23:47:13.202024407Z
 imports:
 - name: github.com/fsnotify/fsnotify
   version: 4da3e2cfbabc9f751898f250b49f2439785783a1
 - name: github.com/go-kit/kit
-  version: 0873e56b0faeae3a1d661b10d629135508ea5504
+  version: 53f10af5d5c7375d4655a3d6852457ed17ab5cc7
   subpackages:
   - log
   - log/level
@@ -12,17 +12,17 @@ imports:
 - name: github.com/go-logfmt/logfmt
   version: 390ab7935ee28ec6b286364bba9b4dd6410cb3d5
 - name: github.com/go-playground/locales
-  version: 1e5f1161c6416a5ff48840eb8724a394e48cc534
+  version: e4cbcb5d0652150d40ad0646651076b6bd2be4f6
   subpackages:
   - currency
 - name: github.com/go-playground/universal-translator
   version: 71201497bace774495daed26a3874fd339e0b538
 - name: github.com/go-stack/stack
-  version: 7a2f19628aabfe68f0766b59e74d6315f8347d22
+  version: 259ab82a6cad3992b4e21ff5cac294ccb06474bc
 - name: github.com/golang/snappy
   version: 553a641470496b2327abcac10b36396bd98e45c9
 - name: github.com/hashicorp/hcl
-  version: a4b07c25de5ff55ad3b8936cea69a79a3d95a855
+  version: 23c074d0eceb2b8a5bfdbb271ab780cde70f05a8
   subpackages:
   - hcl/ast
   - hcl/parser
@@ -39,35 +39,33 @@ imports:
 - name: github.com/kr/logfmt
   version: b84e30acd515aadc4b783ad4ff83aff3299bdfe0
 - name: github.com/magiconair/properties
-  version: 51463bfca2576e06c62a8504b5c0f06d61312647
+  version: 49d762b9817ba1c2e9d0c69183c2b4a8b8f1d934
 - name: github.com/mattn/go-colorable
-  version: ded68f7a9561c023e790de24279db7ebf473ea80
+  version: 6fcc0c1fd9b620311d821b106a400b35dc95c497
 - name: github.com/mattn/go-isatty
-  version: fc9e8d8ef48496124e79ae0df75490096eccf6fe
+  version: 6ca4dbf54d38eea1a992b3c722a76a5d1c4cb25c
 - name: github.com/mitchellh/mapstructure
-  version: cc8532a8e9a55ea36402aa21efdf403a60d34096
-- name: github.com/pelletier/go-buffruneio
-  version: c37440a7cf42ac63b919c752ca73a85067e05992
+  version: 06020f85339e21b2478f756a78e295255ffa4d6a
 - name: github.com/pelletier/go-toml
-  version: 97253b98df84f9eef872866d079e74b8265150f1
+  version: 4e9e0ee19b60b13eb79915933f44d8ed5f268bdd
 - name: github.com/pkg/errors
-  version: c605e284fe17294bda444b34710735b29d1a9d90
+  version: f15c970de5b76fac0b59abb32d62c17cc7bed265
 - name: github.com/spf13/afero
-  version: 9be650865eab0c12963d8753212f4f9c66cdcf12
+  version: 8d919cbe7e2627e417f3e45c3c0e489a5b7e2536
   subpackages:
   - mem
 - name: github.com/spf13/cast
   version: acbeb36b902d72a7a4c18e8f3241075e7ab763e4
 - name: github.com/spf13/cobra
-  version: db6b9a8b3f3f400c8ecb4a4d7d02245b8facad66
+  version: de2d9c4eca8f3c1de17d48b096b6504e0296f003
 - name: github.com/spf13/jwalterweatherman
-  version: fa7ca7e836cf3a8bb4ebf799f472c12d7e903d66
+  version: 12bd96e66386c1960ab0f74ced1362f66f552f7b
 - name: github.com/spf13/pflag
-  version: 80fe0fb4eba54167e2ccae1c6c950e72abf61b73
+  version: 4c012f6dcd9546820e378d0bdda4d8fc772cdfea
 - name: github.com/spf13/viper
-  version: 0967fc9aceab2ce9da34061253ac10fb99bba5b2
+  version: 4dddf7c62e16bce5807744018f5b753bfe21bbd2
 - name: github.com/syndtr/goleveldb
-  version: 8c81ea47d4c41a385645e133e15510fc6a2a74b4
+  version: adf24ef3f94bd13ec4163060b21a5678f22b429b
   subpackages:
   - leveldb
   - leveldb/cache
@@ -82,7 +80,7 @@ imports:
   - leveldb/table
   - leveldb/util
 - name: github.com/tendermint/go-wire
-  version: b53add0b622662731985485f3a19be7f684660b8
+  version: 2baffcb6b690057568bc90ef1d457efb150b979a
   subpackages:
   - data
   - data/base58
@@ -91,25 +89,25 @@ imports:
   subpackages:
   - term
 - name: golang.org/x/crypto
-  version: 5a033cc77e57eca05bdb50522851d29e03569cbe
+  version: 94eea52f7b742c7cbe0b03b22f0c4c8631ece122
   subpackages:
   - ripemd160
 - name: golang.org/x/sys
-  version: 9ccfe848b9db8435a24c424abbc07a921adf1df5
+  version: 8b4580aae2a0dd0c231a45d3ccb8434ff533b840
   subpackages:
   - unix
 - name: golang.org/x/text
-  version: 470f45bf29f4147d6fbd7dfd0a02a848e49f5bf4
+  version: 57961680700a5336d15015c8c50686ca5ba362a4
   subpackages:
   - transform
   - unicode/norm
 - name: gopkg.in/go-playground/validator.v9
-  version: d529ee1b0f30352444f507cc6cdac96bfd12decc
+  version: 61caf9d3038e1af346dbf5c2e16f6678e1548364
 - name: gopkg.in/yaml.v2
-  version: cd8b52f8269e0feb286dfeef29f8fe4d5b397e0b
+  version: 287cf08546ab5e7e37d55a84f7ed3fd1db036de5
 testImports:
 - name: github.com/davecgh/go-spew
-  version: 6d212800a42e8ab5c146b8ace3490ee17e5225f9
+  version: 04cdfd42973bb9c8589fd6a731800cf222fde1a9
   subpackages:
   - spew
 - name: github.com/pmezard/go-difflib
@@ -117,7 +115,7 @@ testImports:
   subpackages:
   - difflib
 - name: github.com/stretchr/testify
-  version: 69483b4bd14f5845b5a1e55bca19e954e827f1d0
+  version: 2aa2c176b9dab406a6970f6a55f513e8a8c8b18f
   subpackages:
   - assert
   - require

glide.yaml

@@ -26,7 +26,6 @@ import:
 - package: gopkg.in/go-playground/validator.v9
 testImport:
 - package: github.com/stretchr/testify
-  version: ^1.1.4
   subpackages:
   - assert
   - require

process/process.go

@@ -1,76 +0,0 @@
-package process
-
-import (
-	"fmt"
-	"io"
-	"os"
-	"os/exec"
-	"time"
-)
-
-type Process struct {
-	Label      string
-	ExecPath   string
-	Args       []string
-	Pid        int
-	StartTime  time.Time
-	EndTime    time.Time
-	Cmd        *exec.Cmd        `json:"-"`
-	ExitState  *os.ProcessState `json:"-"`
-	InputFile  io.Reader        `json:"-"`
-	OutputFile io.WriteCloser   `json:"-"`
-	WaitCh     chan struct{}    `json:"-"`
-}
-
-// execPath: command name
-// args: args to command. (should not include name)
-func StartProcess(label string, dir string, execPath string, args []string, inFile io.Reader, outFile io.WriteCloser) (*Process, error) {
-	cmd := exec.Command(execPath, args...)
-	cmd.Dir = dir
-	cmd.Stdout = outFile
-	cmd.Stderr = outFile
-	cmd.Stdin = inFile
-	if err := cmd.Start(); err != nil {
-		return nil, err
-	}
-	proc := &Process{
-		Label:      label,
-		ExecPath:   execPath,
-		Args:       args,
-		Pid:        cmd.Process.Pid,
-		StartTime:  time.Now(),
-		Cmd:        cmd,
-		ExitState:  nil,
-		InputFile:  inFile,
-		OutputFile: outFile,
-		WaitCh:     make(chan struct{}),
-	}
-	go func() {
-		err := proc.Cmd.Wait()
-		if err != nil {
-			// fmt.Printf("Process exit: %v\n", err)
-			if exitError, ok := err.(*exec.ExitError); ok {
-				proc.ExitState = exitError.ProcessState
-			}
-		}
-		proc.ExitState = proc.Cmd.ProcessState
-		proc.EndTime = time.Now() // TODO make this goroutine-safe
-		err = proc.OutputFile.Close()
-		if err != nil {
-			fmt.Printf("Error closing output file for %v: %v\n", proc.Label, err)
-		}
-		close(proc.WaitCh)
-	}()
-	return proc, nil
-}
-
-func (proc *Process) StopProcess(kill bool) error {
-	defer proc.OutputFile.Close()
-	if kill {
-		// fmt.Printf("Killing process %v\n", proc.Cmd.Process)
-		return proc.Cmd.Process.Kill()
-	} else {
-		// fmt.Printf("Stopping process %v\n", proc.Cmd.Process)
-		return proc.Cmd.Process.Signal(os.Interrupt)
-	}
-}

process/util.go

@@ -1,22 +0,0 @@
-package process
-
-import (
-	. "github.com/tendermint/tmlibs/common"
-)
-
-// Runs a command and gets the result.
-func Run(dir string, command string, args []string) (string, bool, error) {
-	outFile := NewBufferCloser(nil)
-	proc, err := StartProcess("", dir, command, args, nil, outFile)
-	if err != nil {
-		return "", false, err
-	}
-
-	<-proc.WaitCh
-
-	if proc.ExitState.Success() {
-		return outFile.String(), true, nil
-	} else {
-		return outFile.String(), false, nil
-	}
-}

pubsub/query/parser_test.go

@@ -83,9 +83,9 @@ func TestParser(t *testing.T) {
 	for _, c := range cases {
 		_, err := query.New(c.query)
 		if c.valid {
-			assert.NoError(t, err, "Query was '%s'", c.query)
+			assert.NoErrorf(t, err, "Query was '%s'", c.query)
 		} else {
-			assert.Error(t, err, "Query was '%s'", c.query)
+			assert.Errorf(t, err, "Query was '%s'", c.query)
 		}
 	}
 }

pubsub/query/query.go

@@ -22,6 +22,14 @@ type Query struct {
 	parser *QueryParser
 }
 
+// Condition represents a single condition within a query and consists of tag
+// (e.g. "tx.gas"), operator (e.g. "=") and operand (e.g. "7").
+type Condition struct {
+	Tag     string
+	Op      Operator
+	Operand interface{}
+}
+
 // New parses the given string and returns a query or error if the string is
 // invalid.
 func New(s string) (*Query, error) {
@@ -48,17 +56,91 @@ func (q *Query) String() string {
 	return q.str
 }
 
-type operator uint8
+// Operator is an operator that defines some kind of relation between tag and
+// operand (equality, etc.).
+type Operator uint8
 
 const (
-	opLessEqual operator = iota
-	opGreaterEqual
-	opLess
-	opGreater
-	opEqual
-	opContains
+	// "<="
+	OpLessEqual Operator = iota
+	// ">="
+	OpGreaterEqual
+	// "<"
+	OpLess
+	// ">"
+	OpGreater
+	// "="
+	OpEqual
+	// "CONTAINS"; used to check if a string contains a certain sub string.
+	OpContains
 )
 
+// Conditions returns a list of conditions.
+func (q *Query) Conditions() []Condition {
+	conditions := make([]Condition, 0)
+
+	buffer, begin, end := q.parser.Buffer, 0, 0
+
+	var tag string
+	var op Operator
+
+	// tokens must be in the following order: tag ("tx.gas") -> operator ("=") -> operand ("7")
+	for _, token := range q.parser.Tokens() {
+		switch token.pegRule {
+
+		case rulePegText:
+			begin, end = int(token.begin), int(token.end)
+		case ruletag:
+			tag = buffer[begin:end]
+		case rulele:
+			op = OpLessEqual
+		case rulege:
+			op = OpGreaterEqual
+		case rulel:
+			op = OpLess
+		case ruleg:
+			op = OpGreater
+		case ruleequal:
+			op = OpEqual
+		case rulecontains:
+			op = OpContains
+		case rulevalue:
+			// strip single quotes from value (i.e. "'NewBlock'" -> "NewBlock")
+			valueWithoutSingleQuotes := buffer[begin+1 : end-1]
+			conditions = append(conditions, Condition{tag, op, valueWithoutSingleQuotes})
+		case rulenumber:
+			number := buffer[begin:end]
+			if strings.Contains(number, ".") { // if it looks like a floating-point number
+				value, err := strconv.ParseFloat(number, 64)
+				if err != nil {
+					panic(fmt.Sprintf("got %v while trying to parse %s as float64 (should never happen if the grammar is correct)", err, number))
+				}
+				conditions = append(conditions, Condition{tag, op, value})
+			} else {
+				value, err := strconv.ParseInt(number, 10, 64)
+				if err != nil {
+					panic(fmt.Sprintf("got %v while trying to parse %s as int64 (should never happen if the grammar is correct)", err, number))
+				}
+				conditions = append(conditions, Condition{tag, op, value})
+			}
+		case ruletime:
+			value, err := time.Parse(time.RFC3339, buffer[begin:end])
+			if err != nil {
+				panic(fmt.Sprintf("got %v while trying to parse %s as time.Time / RFC3339 (should never happen if the grammar is correct)", err, buffer[begin:end]))
+			}
+			conditions = append(conditions, Condition{tag, op, value})
+		case ruledate:
+			value, err := time.Parse("2006-01-02", buffer[begin:end])
+			if err != nil {
+				panic(fmt.Sprintf("got %v while trying to parse %s as time.Time / '2006-01-02' (should never happen if the grammar is correct)", err, buffer[begin:end]))
+			}
+			conditions = append(conditions, Condition{tag, op, value})
+		}
+	}
+
+	return conditions
+}
+
 // Matches returns true if the query matches the given set of tags, false otherwise.
 //
 // For example, query "name=John" matches tags = {"name": "John"}. More
@@ -71,7 +153,7 @@ func (q *Query) Matches(tags map[string]interface{}) bool {
 	buffer, begin, end := q.parser.Buffer, 0, 0
 
 	var tag string
-	var op operator
+	var op Operator
 
 	// tokens must be in the following order: tag ("tx.gas") -> operator ("=") -> operand ("7")
 	for _, token := range q.parser.Tokens() {
@@ -82,17 +164,17 @@ func (q *Query) Matches(tags map[string]interface{}) bool {
 		case ruletag:
 			tag = buffer[begin:end]
 		case rulele:
-			op = opLessEqual
+			op = OpLessEqual
 		case rulege:
-			op = opGreaterEqual
+			op = OpGreaterEqual
 		case rulel:
-			op = opLess
+			op = OpLess
 		case ruleg:
-			op = opGreater
+			op = OpGreater
 		case ruleequal:
-			op = opEqual
+			op = OpEqual
 		case rulecontains:
-			op = opContains
+			op = OpContains
 		case rulevalue:
 			// strip single quotes from value (i.e. "'NewBlock'" -> "NewBlock")
 			valueWithoutSingleQuotes := buffer[begin+1 : end-1]
@@ -149,7 +231,7 @@ func (q *Query) Matches(tags map[string]interface{}) bool {
 // value from it to the operand using the operator.
 //
 // "tx.gas", "=", "7", { "tx.gas": 7, "tx.ID": "4AE393495334" }
-func match(tag string, op operator, operand reflect.Value, tags map[string]interface{}) bool {
+func match(tag string, op Operator, operand reflect.Value, tags map[string]interface{}) bool {
 	// look up the tag from the query in tags
 	value, ok := tags[tag]
 	if !ok {
@@ -163,15 +245,15 @@ func match(tag string, op operator, operand reflect.Value, tags map[string]inter
 			return false
 		}
 		switch op {
-		case opLessEqual:
+		case OpLessEqual:
 			return v.Before(operandAsTime) || v.Equal(operandAsTime)
-		case opGreaterEqual:
+		case OpGreaterEqual:
 			return v.Equal(operandAsTime) || v.After(operandAsTime)
-		case opLess:
+		case OpLess:
 			return v.Before(operandAsTime)
-		case opGreater:
+		case OpGreater:
 			return v.After(operandAsTime)
-		case opEqual:
+		case OpEqual:
 			return v.Equal(operandAsTime)
 		}
 	case reflect.Float64:
@@ -197,15 +279,15 @@ func match(tag string, op operator, operand reflect.Value, tags map[string]inter
 			panic(fmt.Sprintf("Incomparable types: %T (%v) vs float64 (%v)", value, value, operandFloat64))
 		}
 		switch op {
-		case opLessEqual:
+		case OpLessEqual:
 			return v <= operandFloat64
-		case opGreaterEqual:
+		case OpGreaterEqual:
 			return v >= operandFloat64
-		case opLess:
+		case OpLess:
 			return v < operandFloat64
-		case opGreater:
+		case OpGreater:
 			return v > operandFloat64
-		case opEqual:
+		case OpEqual:
 			return v == operandFloat64
 		}
 	case reflect.Int64:
@@ -231,15 +313,15 @@ func match(tag string, op operator, operand reflect.Value, tags map[string]inter
 			panic(fmt.Sprintf("Incomparable types: %T (%v) vs int64 (%v)", value, value, operandInt))
 		}
 		switch op {
-		case opLessEqual:
+		case OpLessEqual:
 			return v <= operandInt
-		case opGreaterEqual:
+		case OpGreaterEqual:
 			return v >= operandInt
-		case opLess:
+		case OpLess:
 			return v < operandInt
-		case opGreater:
+		case OpGreater:
 			return v > operandInt
-		case opEqual:
+		case OpEqual:
 			return v == operandInt
 		}
 	case reflect.String:
@@ -248,9 +330,9 @@ func match(tag string, op operator, operand reflect.Value, tags map[string]inter
 			return false
 		}
 		switch op {
-		case opEqual:
+		case OpEqual:
 			return v == operand.String()
-		case opContains:
+		case OpContains:
 			return strings.Contains(v, operand.String())
 		}
 	default:

pubsub/query/query_test.go

@@ -45,15 +45,15 @@ func TestMatches(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		query, err := query.New(tc.s)
+		q, err := query.New(tc.s)
 		if !tc.err {
 			require.Nil(t, err)
 		}
 
 		if tc.matches {
-			assert.True(t, query.Matches(tc.tags), "Query '%s' should match %v", tc.s, tc.tags)
+			assert.True(t, q.Matches(tc.tags), "Query '%s' should match %v", tc.s, tc.tags)
 		} else {
-			assert.False(t, query.Matches(tc.tags), "Query '%s' should not match %v", tc.s, tc.tags)
+			assert.False(t, q.Matches(tc.tags), "Query '%s' should not match %v", tc.s, tc.tags)
 		}
 	}
 }
@@ -62,3 +62,24 @@ func TestMustParse(t *testing.T) {
 	assert.Panics(t, func() { query.MustParse("=") })
 	assert.NotPanics(t, func() { query.MustParse("tm.events.type='NewBlock'") })
 }
+
+func TestConditions(t *testing.T) {
+	txTime, err := time.Parse(time.RFC3339, "2013-05-03T14:45:00Z")
+	require.NoError(t, err)
+
+	testCases := []struct {
+		s          string
+		conditions []query.Condition
+	}{
+		{s: "tm.events.type='NewBlock'", conditions: []query.Condition{query.Condition{Tag: "tm.events.type", Op: query.OpEqual, Operand: "NewBlock"}}},
+		{s: "tx.gas > 7 AND tx.gas < 9", conditions: []query.Condition{query.Condition{Tag: "tx.gas", Op: query.OpGreater, Operand: int64(7)}, query.Condition{Tag: "tx.gas", Op: query.OpLess, Operand: int64(9)}}},
+		{s: "tx.time >= TIME 2013-05-03T14:45:00Z", conditions: []query.Condition{query.Condition{Tag: "tx.time", Op: query.OpGreaterEqual, Operand: txTime}}},
+	}
+
+	for _, tc := range testCases {
+		q, err := query.New(tc.s)
+		require.Nil(t, err)
+
+		assert.Equal(t, tc.conditions, q.Conditions())
+	}
+}

test.sh

@@ -1,8 +1,11 @@
 #!/usr/bin/env bash
-
 set -e
-echo "" > coverage.txt
 
+# run the linter
+make metalinter_test
+
+# run the unit tests with coverage
+echo "" > coverage.txt
 for d in $(go list ./... | grep -v vendor); do
     go test -race -coverprofile=profile.out -covermode=atomic "$d"
     if [ -f profile.out ]; then

version/version.go

@@ -1,3 +1,3 @@
 package version
 
-const Version = "0.4.1"
+const Version = "0.5.0"