<commit_before>//  Copyright (c) 2014 Couchbase, Inc.
//  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 main

import (
	"fmt"
	"io/ioutil"
	"os"
	"sync"
	"time"

	log "github.com/couchbaselabs/clog"
)

type ManagerEventHandlers interface {
	OnRegisterPIndex(pindex *PIndex)
	OnUnregisterPIndex(pindex *PIndex)
}

type Manager struct {
	uuid      string // Unique to every Manager process instance.
	startTime time.Time
	version   string // Our software VERSION.
	cfg       Cfg
	bindAddr  string
	dataDir   string
	server    string // The datasource that cbft will index.
	m         sync.Mutex
	feeds     map[string]Feed
	pindexes  map[string]*PIndex
	plannerCh chan string // Used to kick the planner that there's more work.
	janitorCh chan string // Used to kick the janitor that there's more work.
	meh       ManagerEventHandlers
}

func NewManager(version string, cfg Cfg, bindAddr, dataDir string,
	server string, meh ManagerEventHandlers) *Manager {
	return &Manager{
		uuid:      NewUUID(),
		startTime: time.Now(),
		version:   version,
		cfg:       cfg,
		bindAddr:  bindAddr,
		dataDir:   dataDir,
		server:    server,
		feeds:     make(map[string]Feed),
		pindexes:  make(map[string]*PIndex),
		plannerCh: make(chan string),
		janitorCh: make(chan string),
		meh:       meh,
	}
}

func (mgr *Manager) Start() error {
	// TODO: Write our cbft-ID into the cfg.

	if err := mgr.LoadDataDir(); err != nil {
		return err
	}

	// Save our nodeDef into the Cfg.
	if err := mgr.SaveNodeDef(); err != nil {
		return err
	}

	go mgr.PlannerLoop()
	mgr.plannerCh <- "start"

	go mgr.JanitorLoop()
	mgr.janitorCh <- "start"

	return nil
}

func (mgr *Manager) LoadDataDir() error {
	// walk the data dir and register pindexes
	log.Printf("loading dataDir...")
	dirEntries, err := ioutil.ReadDir(mgr.dataDir)
	if err != nil {
		return fmt.Errorf("error: could not read dataDir: %s, err: %v",
			mgr.dataDir, err)
	}

	for _, dirInfo := range dirEntries {
		path := mgr.dataDir + string(os.PathSeparator) + dirInfo.Name()
		name, ok := mgr.ParsePIndexPath(path)
		if !ok {
			log.Printf("  skipping: %s", dirInfo.Name())
			continue
		}

		log.Printf("  opening pindex: %s", name)
		pindex, err := OpenPIndex(name, path)
		if err != nil {
			log.Printf("error: could not open pindex: %s, err: %v",
				path, err)
			continue
		}

		mgr.RegisterPIndex(pindex)
	}

	return nil
}

func (mgr *Manager) SaveNodeDef() error {
	if mgr.cfg == nil {
		return nil // Occurs during testing.
	}

	nodeDefs, cas, err := CfgGetNodeDefs(mgr.cfg, NODE_DEFS_KNOWN)
	if err != nil {
		return err
	}
	if nodeDefs == nil {
		nodeDefs = NewNodeDefs(mgr.version)
	}
	nodeDef, exists := nodeDefs.NodeDefs[mgr.bindAddr]
	if !exists {
		nodeDef = &NodeDef{
			HostPort:    mgr.bindAddr, // TODO: need FQDN:port instead of ":8095".
			UUID:        mgr.uuid,
			ImplVersion: mgr.version,
		}

		nodeDefs.UUID = NewUUID()
		nodeDefs.NodeDefs[mgr.bindAddr] = nodeDef
		nodeDefs.ImplVersion = mgr.version

		_, err = CfgSetNodeDefs(mgr.cfg, NODE_DEFS_KNOWN, nodeDefs, cas)
		if err != nil {
			return err<commit_msg>parametrized SaveNodeDefs()<commit_after>//  Copyright (c) 2014 Couchbase, Inc.
//  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 main

import (
	"fmt"
	"io/ioutil"
	"os"
	"sync"
	"time"

	log "github.com/couchbaselabs/clog"
)

type ManagerEventHandlers interface {
	OnRegisterPIndex(pindex *PIndex)
	OnUnregisterPIndex(pindex *PIndex)
}

type Manager struct {
	uuid      string // Unique to every Manager process instance.
	startTime time.Time
	version   string // Our software VERSION.
	cfg       Cfg
	bindAddr  string
	dataDir   string
	server    string // The datasource that cbft will index.
	m         sync.Mutex
	feeds     map[string]Feed
	pindexes  map[string]*PIndex
	plannerCh chan string // Used to kick the planner that there's more work.
	janitorCh chan string // Used to kick the janitor that there's more work.
	meh       ManagerEventHandlers
}

func NewManager(version string, cfg Cfg, bindAddr, dataDir string,
	server string, meh ManagerEventHandlers) *Manager {
	return &Manager{
		uuid:      NewUUID(),
		startTime: time.Now(),
		version:   version,
		cfg:       cfg,
		bindAddr:  bindAddr,
		dataDir:   dataDir,
		server:    server,
		feeds:     make(map[string]Feed),
		pindexes:  make(map[string]*PIndex),
		plannerCh: make(chan string),
		janitorCh: make(chan string),
		meh:       meh,
	}
}

func (mgr *Manager) Start() error {
	// TODO: Write our cbft-ID into the cfg.

	if err := mgr.LoadDataDir(); err != nil {
		return err
	}

	// Save our nodeDef into the Cfg.
	if err := mgr.SaveNodeDef(NODE_DEFS_KNOWN); err != nil {
		return err
	}

	go mgr.PlannerLoop()
	mgr.plannerCh <- "start"

	go mgr.JanitorLoop()
	mgr.janitorCh <- "start"

	return nil
}

func (mgr *Manager) LoadDataDir() error {
	// walk the data dir and register pindexes
	log.Printf("loading dataDir...")
	dirEntries, err := ioutil.ReadDir(mgr.dataDir)
	if err != nil {
		return fmt.Errorf("error: could not read dataDir: %s, err: %v",
			mgr.dataDir, err)
	}

	for _, dirInfo := range dirEntries {
		path := mgr.dataDir + string(os.PathSeparator) + dirInfo.Name()
		name, ok := mgr.ParsePIndexPath(path)
		if !ok {
			log.Printf("  skipping: %s", dirInfo.Name())
			continue
		}

		log.Printf("  opening pindex: %s", name)
		pindex, err := OpenPIndex(name, path)
		if err != nil {
			log.Printf("error: could not open pindex: %s, err: %v",
				path, err)
			continue
		}

		mgr.RegisterPIndex(pindex)
	}

	return nil
}

func (mgr *Manager) SaveNodeDef(kind string) error {
	if mgr.cfg == nil {
		return nil // Occurs during testing.
	}

	nodeDefs, cas, err := CfgGetNodeDefs(mgr.cfg, kind)
	if err != nil {
		return err
	}
	if nodeDefs == nil {
		nodeDefs = NewNodeDefs(mgr.version)
	}
	nodeDef, exists := nodeDefs.NodeDefs[mgr.bindAddr]
	if !exists {
		nodeDef = &NodeDef{
			HostPort:    mgr.bindAddr, // TODO: need FQDN:port instead of ":8095".
			UUID:        mgr.uuid,
			ImplVersion: mgr.version,
		}

		nodeDefs.UUID = NewUUID()
		nodeDefs.NodeDefs[mgr.bindAddr] = nodeDef
		nodeDefs.ImplVersion = mgr.version

		_, err = CfgSetNodeDefs(mgr.cfg, kind, nodeDefs, cas)
		if err != nil {
			return err
{'fraction_non_alphanumeric': 0.11342207432057681, 'fraction_numerical': 0.0033277870216306157, 'mean_word_length': 3.280712166172107, 'pattern_counts': {'":': 2, '<': 7, '<?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': '3539473', 'n_tokens_mistral': 2654, 'n_tokens_neox': 2535, 'n_words': 773}