Fixes panic on sending errors to recordset. Resolves #67

client.go

@@ -436,9 +436,7 @@ func (clnt *Client) ScanAll(apolicy *ScanPolicy, namespace string, setName strin
 		for _, node := range nodes {
 			go func(node *Node) {
 				if err := clnt.scanNode(&policy, node, res, namespace, setName, binNames...); err != nil {
-					if _, ok := <-res.Errors; ok {
-						res.Errors <- err
-					}
+					res.sendError(err)
 				}
 			}(node)
 		}
@@ -447,9 +445,7 @@ func (clnt *Client) ScanAll(apolicy *ScanPolicy, namespace string, setName strin
 		go func() {
 			for _, node := range nodes {
 				if err := clnt.scanNode(&policy, node, res, namespace, setName, binNames...); err != nil {
-					if _, ok := <-res.Errors; ok {
-						res.Errors <- err
-					}
+					res.sendError(err)
 					continue
 				}
 			}
@@ -860,7 +856,12 @@ func (clnt *Client) Query(policy *QueryPolicy, statement *Statement) (*Recordset
 		// copy policies to avoid race conditions
 		newPolicy := *policy
 		command := newQueryRecordCommand(node, &newPolicy, statement, recSet)
-		go command.Execute()
+		go func() {
+			err := command.Execute()
+			if err != nil {
+				recSet.sendError(err)
+			}
+		}()
 	}
 
 	return recSet, nil
@@ -888,7 +889,12 @@ func (clnt *Client) QueryNode(policy *QueryPolicy, node *Node, statement *Statem
 	// copy policies to avoid race conditions
 	newPolicy := *policy
 	command := newQueryRecordCommand(node, &newPolicy, statement, recSet)
-	go command.Execute()
+	go func() {
+		err := command.Execute()
+		if err != nil {
+			recSet.sendError(err)
+		}
+	}()
 
 	return recSet, nil
 }

recordset.go

@@ -41,6 +41,8 @@ type Recordset struct {
 
 	active    *AtomicBool
 	cancelled chan struct{}
+
+	chanLock sync.Mutex
 }
 
 // NewRecordset generates a new RecordSet instance.
@@ -115,6 +117,8 @@ func (rcs *Recordset) Close() {
 		// wait till all goroutines are done
 		rcs.wgGoroutines.Wait()
 
+		rcs.chanLock.Lock()
+		defer rcs.chanLock.Unlock()
 		close(rcs.Records)
 		close(rcs.Errors)
 	}
@@ -126,3 +130,11 @@ func (rcs *Recordset) signalEnd() {
 		rcs.Close()
 	}
 }
+
+func (rcs *Recordset) sendError(err error) {
+	if rcs.IsActive() {
+		rcs.chanLock.Lock()
+		defer rcs.chanLock.Unlock()
+		rcs.Errors <- err
+	}
+}

recordset_test.go

@@ -0,0 +1,39 @@
+// Copyright 2013-2015 Aerospike, Inc.
+//
+// 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 aerospike
+
+import (
+	"errors"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+// ALL tests are isolated by SetName and Key, which are 50 random charachters
+var _ = Describe("Recordset test", func() {
+
+	It("must avoid panic on sendError", func() {
+		rs := newRecordset(100, 1)
+
+		rs.sendError(errors.New("Error"))
+		rs.wgGoroutines.Done()
+		rs.Close()
+		rs.sendError(errors.New("Error"))
+
+		Expect(<-rs.Errors).NotTo(BeNil())
+		Expect(<-rs.Errors).To(BeNil())
+	})
+
+})