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/memphis
    
    the code of the file until where you have to start completion: package memphis

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

	"github.com/memphisdev/memphis.go"
	"github.com/pkg/errors"
	uuid "github.com/satori/go.uuid"

	"github.com/batchcorp/plumber-schemas/build/go/protos/opts"
	"github.com/batchcorp/plumber-schemas/build/go/protos/records"

	"github.com/streamdal/plumber/validate"
)

func (m *Memphis) 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")
	}

	args := readOpts.GetMemphis().Args

	consumer, err := m.client.CreateConsumer(args.Station, args.ConsumerName, memphis.BatchSize(1))
	if err != nil {
		return errors.Wrap(err, "unable to create Memphis consumer")
	}

	defer consumer.StopConsume()

	var count int64

	msg
    
func (m *Memphis) 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")
	}

	args := readOpts.GetMemphis().Args

	consumer, err := m.client.CreateConsumer(args.Station, args.ConsumerName, memphis.BatchSize(1))
	if err != nil {
		return errors.Wrap(err, "unable to create Memphis consumer")
	}

	defer consumer.StopConsume()

	var count int64

	msgChan := make(chan *memphis.Msg, 0)

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

	consumer.SetContext(ctx)
	handler := getHandler(msgChan, errorChan)
	if err := consumer.Consume(handler); err != nil {
		return errors.Wrap(err, "unable to consume messages")
	}

	// Read forever unless plumber is terminated or --continuous is not specified
	for {
		select {
		case <-ctx.Done():
			return nil
		case msg := <-msgChan:
			count++
			ts := time.Now().UTC().Unix()

			serializedMsg, err := json.Marshal(msg)
			if err != nil {
				errorChan <- &records.ErrorRecord{
					OccurredAtUnixTsUtc: ts,
					Error:               errors.Wrap(err, "unable to serialize message into JSON").Error(),
				}
			}

			resultsChan <- &records.ReadRecord{
				MessageId:           uuid.NewV4().String(),
				Num:                 count,
				Metadata:            msg.GetHeaders(),
				ReceivedAtUnixTsUtc: ts,
				Payload:             msg.Data(),
				XRaw:                serializedMsg,
				Record: &records.ReadRecord_Memphis{
					Memphis: &records.Memphis{
						Value:     msg.Data(),
						Timestamp: ts,
					},
				},
			}

			if !readOpts.Continuous {
				return nil
			}
		}
	}

	return nil
}