<commit_before>package mbus

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/cloudfoundry/yagnats"
	"net/url"
	"os"
	"os/signal"
	"syscall"

	bosherr "github.com/cloudfoundry/bosh-agent/errors"
	boshhandler "github.com/cloudfoundry/bosh-agent/handler"
	boshlog "github.com/cloudfoundry/bosh-agent/logger"
	boshsettings "github.com/cloudfoundry/bosh-agent/settings"
)

const (
	natsHandlerLogTag = "NATS Handler"
	responseMaxLength = 1024 * 1024
)

type natsHandler struct {
	settingsService boshsettings.Service
	client          yagnats.NATSClient
	logger          boshlog.Logger
	handlerFuncs    []boshhandler.Func
}

func NewNatsHandler(
	settingsService boshsettings.Service,
	client yagnats.NATSClient,
	logger boshlog.Logger,
) *natsHandler {
	return &natsHandler{
		settingsService: settingsService,
		client:          client,
		logger:          logger,
	}
}

func (h *natsHandler) Run(handlerFunc boshhandler.Func) error {
	err := h.Start(handlerFunc)
	if err != nil {
		return bosherr.WrapError(err, "Starting nats handler")
	}

	defer h.Stop()

	h.runUntilInterrupted()

	return nil
}

func (h *natsHandler) Start(handlerFunc boshhandler.Func) error {
	h.RegisterAdditionalFunc(handlerFunc)

	connProvider, err := h.getConnectionInfo()
	if err != nil {
		return bosherr.WrapError(err, "Getting connection info")
	}

	err = h.client.Connect(connProvider)
	if err != nil {
		return bosherr.WrapError(err, "Connecting")
	}

	settings := h.settingsService.GetSettings()

	subject := fmt.Sprintf("agent.%s", settings.AgentID)

	h.logger.Info(natsHandlerLogTag, "Subscribing to %s", subject)

	_, err = h.client.Subscribe(subject, func(natsMsg *yagnats.Message) {
		for _, handlerFunc := range h.handlerFuncs {
			h.handleNatsMsg(natsMsg, handlerFunc)
		}
	})

	return nil
}

func (h *natsHandler) RegisterAdditionalFunc(handlerFunc boshhandler.Func) {
	// Currently not locking since RegisterAdditionalFunc
	// is not a primary way of adding handlerFunc.
	h.handlerFuncs = append(h.handlerFuncs, handlerFunc)<commit_msg>Added logging for NATS subscription failure<commit_after>package mbus

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/cloudfoundry/yagnats"
	"net/url"
	"os"
	"os/signal"
	"syscall"

	bosherr "github.com/cloudfoundry/bosh-agent/errors"
	boshhandler "github.com/cloudfoundry/bosh-agent/handler"
	boshlog "github.com/cloudfoundry/bosh-agent/logger"
	boshsettings "github.com/cloudfoundry/bosh-agent/settings"
)

const (
	natsHandlerLogTag = "NATS Handler"
	responseMaxLength = 1024 * 1024
)

type natsHandler struct {
	settingsService boshsettings.Service
	client          yagnats.NATSClient
	logger          boshlog.Logger
	handlerFuncs    []boshhandler.Func
}

func NewNatsHandler(
	settingsService boshsettings.Service,
	client yagnats.NATSClient,
	logger boshlog.Logger,
) *natsHandler {
	return &natsHandler{
		settingsService: settingsService,
		client:          client,
		logger:          logger,
	}
}

func (h *natsHandler) Run(handlerFunc boshhandler.Func) error {
	err := h.Start(handlerFunc)
	if err != nil {
		return bosherr.WrapError(err, "Starting nats handler")
	}

	defer h.Stop()

	h.runUntilInterrupted()

	return nil
}

func (h *natsHandler) Start(handlerFunc boshhandler.Func) error {
	h.RegisterAdditionalFunc(handlerFunc)

	connProvider, err := h.getConnectionInfo()
	if err != nil {
		return bosherr.WrapError(err, "Getting connection info")
	}

	err = h.client.Connect(connProvider)
	if err != nil {
		return bosherr.WrapError(err, "Connecting")
	}

	settings := h.settingsService.GetSettings()

	subject := fmt.Sprintf("agent.%s", settings.AgentID)

	h.logger.Info(natsHandlerLogTag, "Subscribing to %s", subject)

	_, err = h.client.Subscribe(subject, func(natsMsg *yagnats.Message) {
		for _, handlerFunc := range h.handlerFuncs {
			h.handleNatsMsg(natsMsg, handlerFunc)
		}
	})

	if err != nil {
		h.logger.Error(natsHandlerLogTag, bosherr.WrapError(err, "Subscribing to %s", subject).Error())
	}

	return nil
}

func (h *natsHandler) RegisterAdditionalFunc(handlerFunc boshhandler.Func) {
	// Currently not locking since RegisterAdditionalFunc
	// is not a primary way of adding handlerFunc.
	h.handlerFuncs = append(h.handlerFuncs, handlerFunc)
{'fraction_non_alphanumeric': 0.11298362212200333, 'fraction_numerical': 0.0037977688108236413, 'mean_word_length': 4.465629053177691, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5355005', 'n_tokens_mistral': 1536, 'n_tokens_neox': 1446, 'n_words': 301}