func (c *Manager) ProcessMessages(msgs []*fspb.Message) {
	ctx, fin := context.WithTimeout(context.Background(), 30*time.Second)
	hasResult := make([]bool, len(msgs))
	var working sync.WaitGroup
	working.Add(len(msgs))
	for idx, msg := range msgs {
		i, m := idx, msg
		c.stats.MessageIngested(true, m)
		go func() {
			defer working.Done()
			l := c.services[m.Destination.ServiceName]
			if l == nil {
				log.Errorf("Message in datastore [%v] is for unknown service [%s].", hex.EncodeToString(m.MessageId), m.Destination.ServiceName)
				return
			}
			res := l.processMessage(ctx, m)
			if res != nil {
				hasResult[i] = true
				m.Result = res
			}
		}()
	}
	working.Wait()
	fin()
	toSave := make([]*fspb.Message, 0, len(msgs))
	for i, m := range msgs {
		if hasResult[i] {
			toSave = append(toSave, m)
		}
	}
	if len(toSave) == 0 {
		return
	}
	ctx, fin = context.WithTimeout(context.Background(), 15*time.Second)
	defer fin()
	if err := c.dataStore.StoreMessages(ctx, toSave, ""); err != nil {
		log.Errorf("Error saving results for %d messages: %v", len(toSave), err)
	}
}
ProcessMessages implements MessageProcessor and is called by the datastore on
backlogged messages.
package services
import (
	"context"
	"encoding/hex"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	log "github.com/golang/glog"
	"github.com/golang/protobuf/proto"
	"golang.org/x/time/rate"
	"github.com/google/fleetspeak/fleetspeak/src/common"
	"github.com/google/fleetspeak/fleetspeak/src/server/db"
	"github.com/google/fleetspeak/fleetspeak/src/server/internal/cache"
	"github.com/google/fleetspeak/fleetspeak/src/server/internal/ftime"
	"github.com/google/fleetspeak/fleetspeak/src/server/service"
	"github.com/google/fleetspeak/fleetspeak/src/server/stats"
	fspb "github.com/google/fleetspeak/fleetspeak/src/common/proto/fleetspeak"
	spb "github.com/google/fleetspeak/fleetspeak/src/server/proto/fleetspeak_server"
)
const MaxServiceFailureReasonLength = 900
type Manager struct {
	services        map[string]*liveService
	dataStore       db.Store
	serviceRegistry map[string]service.Factory  
	stats           stats.Collector
	cc              *cache.Clients
}
func NewManager(dataStore db.Store, serviceRegistry map[string]service.Factory, stats stats.Collector, clientCache *cache.Clients) *Manager {
	m := Manager{
		services:        make(map[string]*liveService),
		dataStore:       dataStore,
		serviceRegistry: serviceRegistry,
		stats:           stats,
		cc:              clientCache,
	}
	ssd := liveService{
		manager:        &m,
		name:           "system",
		maxParallelism: 100,
		pLogLimiter:    rate.NewLimiter(rate.Every(10*time.Second), 1),
	}
	ss := systemService{
		sctx:      &ssd,
		stats:     stats,
		datastore: dataStore,
		cc:        clientCache,
	}
	ssd.service = &ss
	m.services["system"] = &ssd
	ss.Start(&ssd)
	return &m
}
func (c *Manager) Install(cfg *spb.ServiceConfig) error {
	cfg = proto.Clone(cfg).(*spb.ServiceConfig)
	f := c.serviceRegistry[cfg.Factory]
	if f == nil {
		return fmt.Errorf("unable to find factory [%v]", cfg.Factory)
	}
	if cfg.Name == "" || cfg.Name == "system" || cfg.Name == "client" {
		return fmt.Errorf("illegal service name [%v]", cfg.Name)
	}
	s, err := f(cfg)
	if err != nil {
		return err
	}
	if cfg.MaxParallelism == 0 {
		cfg.MaxParallelism = 100
	}
	d := liveService{
		manager: c,
		name:    cfg.Name,
		service: s,
		maxParallelism: cfg.MaxParallelism,
		pLogLimiter:    rate.NewLimiter(rate.Every(10*time.Second), 1),
	}
	if err = s.Start(&d); err != nil {
		return err
	}
	c.services[cfg.Name] = &d
	log.Infof("Installed %v service.", cfg.Name)
	return nil
}
func (c *Manager) Stop() {
	for _, d := range c.services {
		d.stop()
	}
	c.services = map[string]*liveService{}
}