recorder concurrency support using channels.

Author: tonyredondo

agent/recorder.go

@@ -101,7 +101,6 @@ func NewSpanRecorder(agent *Agent) *SpanRecorder {
 	r.stats = &RecorderStats{}
 	r.logger.Printf("recorder frequency: %v", agent.flushFrequency)
 	r.logger.Printf("recorder concurrency level: %v", agent.concurrencyLevel)
-
 	// start workers
 	r.workerJobs = make(chan *workerJob, r.concurrencyLevel)
 	r.workerResults = make(chan *workerResult, r.concurrencyLevel)
@@ -129,8 +128,18 @@ func (r *SpanRecorder) RecordSpan(span tracer.RawSpan) {
 
 func (r *SpanRecorder) loop() error {
 	defer func() {
+		close(r.workerJobs)
+		close(r.workerResults)
 		r.logger.Println("recorder has been stopped.")
 	}()
+
+	// start workers
+	r.workerJobs = make(chan *workerJob, r.concurrencyLevel)
+	r.workerResults = make(chan *workerResult, r.concurrencyLevel)
+	for i := 0; i < r.concurrencyLevel; i++ {
+		go r.worker(i + 1)
+	}
+
 	ticker := time.NewTicker(1 * time.Second)
 	cTime := time.Now()
 	for {