func (p *Processor) GetWorkers() []*worker.Worker {
	var workers []*worker.Worker
	for _, triggerInstance := range p.triggers {
		workers = append(workers, triggerInstance.GetWorkers()...)
	}
	return workers
}
GetWorkers returns workers
package app
import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sync"
	"time"
	"github.com/nuclio/nuclio/pkg/common"
	commonhealthcheck "github.com/nuclio/nuclio/pkg/common/healthcheck"
	"github.com/nuclio/nuclio/pkg/common/status"
	"github.com/nuclio/nuclio/pkg/errgroup"
	"github.com/nuclio/nuclio/pkg/functionconfig"
	"github.com/nuclio/nuclio/pkg/loggersink"
	"github.com/nuclio/nuclio/pkg/platform/abstract"
	"github.com/nuclio/nuclio/pkg/platformconfig"
	"github.com/nuclio/nuclio/pkg/processor"
	"github.com/nuclio/nuclio/pkg/processor/config"
	"github.com/nuclio/nuclio/pkg/processor/healthcheck"
	"github.com/nuclio/nuclio/pkg/processor/metricsink"
	"github.com/nuclio/nuclio/pkg/processor/runtime"
	_ "github.com/nuclio/nuclio/pkg/processor/runtime/dotnetcore"
	_ "github.com/nuclio/nuclio/pkg/processor/runtime/golang"
	_ "github.com/nuclio/nuclio/pkg/processor/runtime/java"
	_ "github.com/nuclio/nuclio/pkg/processor/runtime/nodejs"
	_ "github.com/nuclio/nuclio/pkg/processor/runtime/python"
	_ "github.com/nuclio/nuclio/pkg/processor/runtime/ruby"
	_ "github.com/nuclio/nuclio/pkg/processor/runtime/shell"
	"github.com/nuclio/nuclio/pkg/processor/timeout"
	"github.com/nuclio/nuclio/pkg/processor/trigger"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/cron"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/http"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/kafka"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/kickstart"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/kinesis"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/mqtt/basic"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/mqtt/iotcore"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/nats"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/partitioned/eventhub"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/poller/v3ioitempoller"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/pubsub"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/rabbitmq"
	_ "github.com/nuclio/nuclio/pkg/processor/trigger/v3iostream"
	"github.com/nuclio/nuclio/pkg/processor/util/clock"
	"github.com/nuclio/nuclio/pkg/processor/webadmin"
	"github.com/nuclio/nuclio/pkg/processor/worker"
	_ "github.com/nuclio/nuclio/pkg/sinks"
	"github.com/nuclio/errors"
	"github.com/nuclio/logger"
	"github.com/v3io/version-go"
)
type Processor struct {
	logger                logger.Logger
	functionLogger        logger.Logger
	triggers              []trigger.Trigger
	webAdminServer        *webadmin.Server
	healthCheckServer     commonhealthcheck.Server
	metricSinks           []metricsink.MetricSink
	namedWorkerAllocators *worker.AllocatorSyncMap
	eventTimeoutWatcher   *timeout.EventTimeoutWatcher
	startComplete         bool
	stop                  chan bool
}
func NewProcessor(configurationPath string, platformConfigurationPath string) (*Processor, error) {
	var err error
	newProcessor := &Processor{
		namedWorkerAllocators: worker.NewAllocatorSyncMap(),
		stop:                  make(chan bool, 1),
	}
	platformConfiguration, err := platformconfig.NewPlatformConfig(platformConfigurationPath)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to get platform configuration")
	}
	processorConfiguration, err := newProcessor.readConfiguration(configurationPath)
	if err != nil {
		return nil, err
	}
	newProcessor.logger, err = loggersink.CreateFunctionLogger("processor",
		&processorConfiguration.Config,
		platformConfiguration)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to create logger")
	}
	newProcessor.functionLogger = newProcessor.logger
	newProcessor.logger.InfoWith("Starting processor", "version", version.Get())
	indentedProcessorConfiguration, _ := json.MarshalIndent(processorConfiguration, "", "    ")
	indentedPlatformConfiguration, _ := json.MarshalIndent(platformConfiguration, "", "    ")
	newProcessor.logger.DebugWith("Read configuration",
		"config", string(indentedProcessorConfiguration),
		"platformConfig", string(indentedPlatformConfiguration))
	processorConfiguration.PlatformConfig = platformConfiguration
	if processorConfiguration.Spec.EventTimeout != "" {
		clock.SetResolution(1 * time.Second)
	}
	newProcessor.healthCheckServer, err = newProcessor.createAndStartHealthCheckServer(platformConfiguration)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to create and start health check server")
	}
	newProcessor.triggers, err = newProcessor.createTriggers(processorConfiguration)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to create triggers")
	}
	if len(processorConfiguration.Spec.EventTimeout) > 0 {
		eventTimeout, timeoutErr := processorConfiguration.Spec.GetEventTimeout()
		if timeoutErr != nil {
			return nil, errors.Wrap(timeoutErr, "Bad EventTimeout")
		}
		if startErr := newProcessor.startTimeoutWatcher(eventTimeout); startErr != nil {
			return nil, errors.Wrap(startErr, "Can't start timeout watcher")
		}
	}
	newProcessor.webAdminServer, err = newProcessor.createWebAdminServer(platformConfiguration)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to create web interface server")
	}
	newProcessor.metricSinks, err = newProcessor.createMetricSinks(processorConfiguration, platformConfiguration)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to create metric sinks")
	}
	return newProcessor, nil
}
func (p *Processor) Start() error {
	p.logger.DebugWith("Starting triggers", "triggers", p.triggers)
	for _, triggerInstance := range p.triggers {
		if err := triggerInstance.Start(nil); err != nil {
			p.logger.ErrorWith("Failed to start trigger",
				"kind", triggerInstance.GetKind(),
				"err", err.Error())
			return errors.Wrap(err, "Failed to start trigger")
		}
	}
	if err := p.webAdminServer.Start(); err != nil {
		return errors.Wrap(err, "Failed to start web interface")
	}
	for _, metricPusher := range p.metricSinks {
		if err := metricPusher.Start(); err != nil {
			return errors.Wrap(err, "Failed to start metric pushing")
		}
	}
	p.startComplete = true
	p.logger.Debug("Processor started")
	<-p.stop  
	p.logger.Info("Processor quitting")
	time.Sleep(5 * time.Second)  
	return nil
}
func (p *Processor) GetTriggers() []trigger.Trigger {
	return p.triggers
}