<commit_before>// Copyright 2016 IBM Corporation
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//       http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.

package checker

import (
	"bytes"

	"encoding/json"

	"github.com/Sirupsen/logrus"
	"github.com/amalgam8/controller/resources"
	"github.com/amalgam8/sidecar/config"
	"github.com/amalgam8/sidecar/router/clients"
	"github.com/amalgam8/sidecar/router/nginx"
)

// Listener listens for events from Message Hub and updates NGINX
type Listener interface {
	Start() error
	Stop() error
}

type listener struct {
	config     *config.Config
	consumer   Consumer
	nginx      nginx.Nginx
	controller clients.Controller
}

// NewListener new Listener implementation
func NewListener(config *config.Config, consumer Consumer, c clients.Controller, nginx nginx.Nginx) Listener {
	return &listener{
		config:     config,
		consumer:   consumer,
		nginx:      nginx,
		controller: c,
	}
}

// Start listens messages to arrive
func (l *listener) Start() error {
	logrus.Info("Listening for messages")
	for {
		err := l.listenForUpdate()
		if err != nil {
			logrus.WithError(err).Error("Update failed")
		}
	}
}

// ListenForUpdate sleeps until an event indicating that the rules for this tenant have
// changed. Once the event occurs we attempt to update our configuration.
func (l *listener) listenForUpdate() error {

	confTemplate := resources.ConfigTemplate{}

	// Sleep until we receive an event indicating that the our rules have changed
	for {
		key, value, err := l.consumer.ReceiveEvent()
		if err != nil {
			logrus.WithError(err).Error("Couldn't read from Kafka bus")
			return err
		}

		if key == l.config.Tenant.Token {
			err = json.Unmarshal(value, &confTemplate)
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"err":   err,
					"key":   key,
					"value": string(value),
				}).Error("Received message was not in expected format")
			} else {

				logrus.WithFields(logrus.Fields{
					"key":   key,
					"value": confTemplate,
				}).Info("Tenant event received")
				break
			}
		}
	}

	// TODO
	// use configTemplate object

	// Get latest config from Controller
	conf, err := l.controller.GetNGINXConfig(nil)
	if err != nil {
		logrus.WithError(err).Error("Call to Controller failed")
		return err
	}
	//since version="", Controller should never return empty string

	reader := bytes.NewBufferString(conf)
<commit_msg>removing unmarshalling of byte array to controller defined object<commit_after>// Copyright 2016 IBM Corporation
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//       http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.

package checker

import (
	"bytes"

	"github.com/Sirupsen/logrus"
	"github.com/amalgam8/sidecar/config"
	"github.com/amalgam8/sidecar/router/clients"
	"github.com/amalgam8/sidecar/router/nginx"
)

// Listener listens for events from Message Hub and updates NGINX
type Listener interface {
	Start() error
	Stop() error
}

type listener struct {
	config     *config.Config
	consumer   Consumer
	nginx      nginx.Nginx
	controller clients.Controller
}

// NewListener new Listener implementation
func NewListener(config *config.Config, consumer Consumer, c clients.Controller, nginx nginx.Nginx) Listener {
	return &listener{
		config:     config,
		consumer:   consumer,
		nginx:      nginx,
		controller: c,
	}
}

// Start listens messages to arrive
func (l *listener) Start() error {
	logrus.Info("Listening for messages")
	for {
		err := l.listenForUpdate()
		if err != nil {
			logrus.WithError(err).Error("Update failed")
		}
	}
}

// ListenForUpdate sleeps until an event indicating that the rules for this tenant have
// changed. Once the event occurs we attempt to update our configuration.
func (l *listener) listenForUpdate() error {

	// Sleep until we receive an event indicating that the our rules have changed
	for {
		key, value, err := l.consumer.ReceiveEvent()
		if err != nil {
			logrus.WithError(err).Error("Couldn't read from Kafka bus")
			return err
		}

		if key == l.config.Tenant.Token {
			logrus.WithFields(logrus.Fields{
				"key":   key,
				"value": string(value),
			}).Info("Tenant event received")
			break
		}
	}

	// TODO
	// use configTemplate object

	// Get latest config from Controller
	conf, err := l.controller.GetNGINXConfig(nil)
	if err != nil {
		logrus.WithError(err).Error("Call to Controller failed")
		return err
	}
	//since version="", Controller should never return empty string

	reader := bytes.NewBufferString(conf)

{'fraction_non_alphanumeric': 0.10269174041297935, 'fraction_numerical': 0.004240412979351032, 'mean_word_length': 3.4503691550451188, 'pattern_counts': {'":': 7, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2457764', 'n_tokens_mistral': 1752, 'n_tokens_neox': 1609, 'n_words': 561}