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: connection.go
    path of file: ./repos/marketstore/contrib/alpaca/api
    
    the code of the file until where you have to start completion: package api

import (
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"github.com/gorilla/websocket"

	"github.com/alpacahq/marketstore/v4/contrib/alpaca/config"
	"github.com/alpacahq/marketstore/v4/utils/log"
)

var errExchangeMessage = errors.New("didn't receive expected message")

type AlpacaWebSocket struct {
	maxMessageSize int64
	pingPeriod     time.Duration
	server         string
	apiKey         string
	apiSecret      string
	subscriptions  []string
	conn           *websocket.Conn
	outputChan     chan<- interface{}
}

func NewAlpacaWebSocket(cfg *config.Config, oChan chan<- interface{}) *AlpacaWebSocket {
	const defaultMaxMessageSizeInBytes = 2048000
	return &AlpacaWebSocket{
		maxMessageSize: defaultMaxMessageSizeInBytes,
		pingPeriod:     10 * time.Second,
		server:         cfg.WSServer,
		apiKey:         cfg.APIKey,
		apiSecret:      cfg.APISecret,
		subscriptions:  cfg.Subscription.AsCanonical(),
		conn:           nil,
		outputChan:
    
func (p *AlpacaWebSocket) listen() error {
	// start the websocket connection
	if err := p.connect(); err != nil {
		log.Error("[alpaca] error connecting to server {%s:%v,%s:%v,%s:%s}",
			"server", p.server,
			"subscription", p.subscriptions,
			"error", err)
		return err
	}
	defer func(conn *websocket.Conn) {
		if err := conn.Close(); err != nil {
			log.Error("failed to close websocket connection", err.Error())
		}
	}(p.conn)

	p.conn.SetReadLimit(p.maxMessageSize)
	p.conn.SetPongHandler(func(string) error {
		// The ping we have sent has received a reply
		// so we extend the deadline beyond our next ping
		return p.setReadDeadline()
	})

	// Subscribe to streams
	if err := p.subscribe(); err != nil {
		return err
	}

	// errorChan is buffered to ensure receiveMessages
	// can always finish
	out, errorChan := make(chan []byte), make(chan error, 1)
	go p.receiveMessages(out, errorChan)
	ticker := time.NewTicker(p.pingPeriod)
	defer ticker.Stop()

	for {
		select {
		case err := <-errorChan:
			return err
		case <-ticker.C:
			err := p.conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
			if err != nil {
				log.Error("[alpaca] stream write ping error %s", err)
				return err
			}
		case msg := <-out:
			p.outputChan <- msg
		}
	}
}