You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: read.go
    path of file: ./repos/plumber/backends/activemq
    
    the code of the file until where you have to start completion: package activemq

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/go-stomp/stomp"
	"github.com/pkg/errors"
	uuid "github.com/satori/go.uuid"

	"github.com/streamdal/plumber/validate"

	"github.com/batchcorp/plumber-schemas/build/go/protos/args"
	
    
func (a *ActiveMQ) Read(ctx context.Context, readOpts *opts.ReadOptions, resultsChan chan *records.ReadRecord, errorChan chan *records.ErrorRecord) error {
	if err := validateReadOptions(readOpts); err != nil {
		return errors.Wrap(err, "invalid read options")
	}

	var count int64

	sub, err := a.client.Subscribe(getDestinationRead(readOpts.Activemq.Args), stomp.AckClient)
	if err != nil {
		return errors.Wrap(err, "unable to create subscription")
	}

	defer sub.Unsubscribe()

	a.log.Info("Listening for message(s) ...")

	for {
		select {
		case <-ctx.Done():
			return nil
		case msg := <-sub.C:
			if msg.Err != nil {
				errorChan <- &records.ErrorRecord{
					Error:               msg.Err.Error(),
					OccurredAtUnixTsUtc: time.Now().UTC().Unix(),
				}
				continue
			}

			count++

			// can marshal entire message: unsupported type: chan *stomp.Message
			serializedMsg, err := json.Marshal(msg.Body)
			if err != nil {
				errorChan <- &records.ErrorRecord{
					OccurredAtUnixTsUtc: time.Now().UTC().Unix(),
					Error:               errors.Wrap(err, "unable to serialize message into JSON").Error(),
				}
			}

			if err := a.client.Ack(msg); err != nil {
				errorChan <- &records.ErrorRecord{
					OccurredAtUnixTsUtc: time.Now().UTC().Unix(),
					Error:               fmt.Sprintf("unable to ack message: %s", err),
				}
			}

			resultsChan <- &records.ReadRecord{
				MessageId:           uuid.NewV4().String(),
				Num:                 count,
				ReceivedAtUnixTsUtc: time.Now().UTC().Unix(),
				Payload:             msg.Body,
				XRaw:                serializedMsg,
				Record: &records.ReadRecord_Activemq{
					Activemq: &records.ActiveMQ{
						Destination:    msg.Destination,
						ContentType:    msg.ContentType,
						SubscriptionId: msg.Subscription.Id(),
						Value:          msg.Body,
					},
				},
			}

			if !readOpts.Continuous {
				return nil
			}
		}
	}

	return nil
}