<commit_before>/**********************************************************************************
* Copyright (c) 2009-2017 Misakai Ltd.
* This program is free software: you can redistribute it and/or modify it under the
* terms of the GNU Affero General Public License as published by the  Free Software
* Foundation, either version 3 of the License, or(at your option) any later version.
*
* This program is distributed  in the hope that it  will be useful, but WITHOUT ANY
* WARRANTY;  without even  the implied warranty of MERCHANTABILITY or FITNESS FOR A
* PARTICULAR PURPOSE.  See the GNU Affero General Public License  for  more details.
*
* You should have  received a copy  of the  GNU Affero General Public License along
* with this program. If not, see<http://www.gnu.org/licenses/>.
************************************************************************************/

package broker

import (
	"fmt"
	"net"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/emitter-io/emitter/config"
	"github.com/emitter-io/emitter/logging"
	"github.com/emitter-io/emitter/network/address"
	"github.com/emitter-io/emitter/network/listener"
	"github.com/emitter-io/emitter/network/tcp"
	"github.com/emitter-io/emitter/network/websocket"
	"github.com/emitter-io/emitter/perf"
	"github.com/emitter-io/emitter/security"
	"github.com/hashicorp/memberlist"
	"github.com/hashicorp/serf/serf"
)

// Service represents the main structure.
type Service struct {
	Closing          chan bool                 // The channel for closing signal.
	Counters         *perf.Counters            // The performance counters for this service.
	Cipher           *security.Cipher          // The cipher to use for decoding and encoding keys.
	License          *security.License         // The licence for this emitter server.
	Config           *config.Config            // The configuration for the service.
	ContractProvider security.ContractProvider // The contract provider for the service.
	subscriptions    *SubscriptionTrie         // The subscription matching trie.
	http             *http.Server              // The underlying HTTP server.
	tcp              *tcp.Server               // The underlying TCP server.
	cluster          *serf.Serf                // The gossip-based cluster mechanism.
	events           chan serf.Event           // The channel for receiving gossip events.
}

// NewService creates a new service.
func NewService(cfg *config.Config) (s *Service, err error) {
	s = &Service{
		Closing:       make(chan bool),
		Counters:      perf.NewCounters(),
		Config:        cfg,
		subscriptions: NewSubscriptionTrie(),
		events:        make(chan serf.Event),
		http:          new(http.Server),
		tcp:           new(tcp.Server),
	}

	// Attach handlers
	s.tcp.Handler = s.onAccept
	http.HandleFunc("/", s.onRequest)

	// Parse the license
	logging.LogAction("service", "external address is "+address.External().String())
	logging.LogAction("service", "reading the license...")
	if s.License, err = security.ParseLicense(cfg.License); err != nil {
		return nil, err
	}

	// Create a new cipher from the licence provided
	if s.Cipher, err = s.License.Cipher(); err != nil {
		return nil, err
	}

	go func() {
		select {
		case e := <-s.events:
			println("Event: " + e.String())
		}
	}()

	// Create the cluster
	if s.cluster, err = serf.Create(s.clusterConfig(cfg)); err != nil {
		return nil, err
	}

	// Hook up signal handling<commit_msg>do not create cluster gossip if not configured<commit_after>/**********************************************************************************
* Copyright (c) 2009-2017 Misakai Ltd.
* This program is free software: you can redistribute it and/or modify it under the
* terms of the GNU Affero General Public License as published by the  Free Software
* Foundation, either version 3 of the License, or(at your option) any later version.
*
* This program is distributed  in the hope that it  will be useful, but WITHOUT ANY
* WARRANTY;  without even  the implied warranty of MERCHANTABILITY or FITNESS FOR A
* PARTICULAR PURPOSE.  See the GNU Affero General Public License  for  more details.
*
* You should have  received a copy  of the  GNU Affero General Public License along
* with this program. If not, see<http://www.gnu.org/licenses/>.
************************************************************************************/

package broker

import (
	"fmt"
	"net"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/emitter-io/emitter/config"
	"github.com/emitter-io/emitter/logging"
	"github.com/emitter-io/emitter/network/address"
	"github.com/emitter-io/emitter/network/listener"
	"github.com/emitter-io/emitter/network/tcp"
	"github.com/emitter-io/emitter/network/websocket"
	"github.com/emitter-io/emitter/perf"
	"github.com/emitter-io/emitter/security"
	"github.com/hashicorp/memberlist"
	"github.com/hashicorp/serf/serf"
)

// Service represents the main structure.
type Service struct {
	Closing          chan bool                 // The channel for closing signal.
	Counters         *perf.Counters            // The performance counters for this service.
	Cipher           *security.Cipher          // The cipher to use for decoding and encoding keys.
	License          *security.License         // The licence for this emitter server.
	Config           *config.Config            // The configuration for the service.
	ContractProvider security.ContractProvider // The contract provider for the service.
	subscriptions    *SubscriptionTrie         // The subscription matching trie.
	http             *http.Server              // The underlying HTTP server.
	tcp              *tcp.Server               // The underlying TCP server.
	cluster          *serf.Serf                // The gossip-based cluster mechanism.
	events           chan serf.Event           // The channel for receiving gossip events.
}

// NewService creates a new service.
func NewService(cfg *config.Config) (s *Service, err error) {
	s = &Service{
		Closing:       make(chan bool),
		Counters:      perf.NewCounters(),
		Config:        cfg,
		subscriptions: NewSubscriptionTrie(),
		events:        make(chan serf.Event),
		http:          new(http.Server),
		tcp:           new(tcp.Server),
	}

	// Attach handlers
	s.tcp.Handler = s.onAccept
	http.HandleFunc("/", s.onRequest)

	// Parse the license
	logging.LogAction("service", "external address is "+address.External().String())
	logging.LogAction("service", "reading the license...")
	if s.License, err = security.ParseLicense(cfg.License); err != nil {
		return nil, err
	}

	// Create a new cipher from the licence provided
	if s.Cipher, err = s.License.Cipher(); err != nil {
		return nil, err
	}

	go func() {
		select {
		case e := <-s.events:
			println("Event: " + e.String())
		}
	}()

	// Create the cluster
	if cfg.Cluster != nil {
		if s.cluster, err = serf.Create(s.clusterConfig(cfg)); err != nil {
			return nil, err
		}
	}

	// Hook up signal handling
{'fraction_non_alphanumeric': 0.14778254649499284, 'fraction_numerical': 0.002575107296137339, 'mean_word_length': 2.9231200897867566, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, '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': '1952487', 'n_tokens_mistral': 2074, 'n_tokens_neox': 1931, 'n_words': 641}