process

README

Process Manager

Installation

go get -u github.com/s-larionov/process-manager

Example

package main

import (
	"context"
	"net/http"
	"os"
	"os/signal"
	"syscall"

	"github.com/gorilla/mux"
	"github.com/prometheus/client_golang/prometheus/promhttp"

	"github.com/s-larionov/process-manager"
)


func main() {
	manager := process.NewManager()

	// Create a callback worker
	manager.AddWorker(process.NewCallbackWorker("test", func(ctx context.Context) error {
		select {
		case <-ctx.Done():
			return ctx.Err()
		}
	}))

	// Create a callback worker with retries
	opt := process.RetryOnErrorOpt{
		Timeout:     time.Second, // When it is omitted manager will try to run it immediately
		MaxAttempts: 10,          // When this param is missed manager will try restart in infinity loop
	}
	manager.AddWorker(process.NewCallbackWorker("test with error", func(ctx context.Context) error {
		for {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(time.Millisecond):
				return errors.New("test error")
			}
		}
	}, opt))

	// Create an example of server worker for prometheus
	handler := mux.NewRouter()
	handler.Handle("/metrics", promhttp.Handler())
	server := &http.Server{
		Addr:    ":2112",
		Handler: handler,
	}
	manager.AddWorker(process.NewServerWorker("prometheus", server))

	manager.StartAll()

	WaitShutdown(manager)
}

func WaitShutdown(manager *process.Manager) {
	go func(manager *process.Manager) {
		sigChan := make(chan os.Signal, 1)
		signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

		<-sigChan

		manager.StopAll()
	}(manager)

	manager.AwaitAll()
}

Documentation

Index

• Variables
• func SetLogger(logger Logger)
• type CallbackFunc
• type CallbackOpt
• type CallbackWorker
  • func NewCallbackWorker(name string, cb CallbackFunc, opts ...CallbackOpt) *CallbackWorker
  • func (w *CallbackWorker) Start() error
  • func (w *CallbackWorker) Stop() error
• type DummyLog
  • func (DummyLog) Error(_ string, _ error, _ ...LogFields)
  • func (DummyLog) Info(_ string, _ ...LogFields)
• type LogFields
• type Logger
• type Manager
  • func NewManager() *Manager
  • func NewManagerWithWorkers(workers []Worker) *Manager
  • func (m *Manager) AddWorker(worker Worker)
  • func (m *Manager) AwaitAll()
  • func (m *Manager) IsRunning() bool
  • func (m *Manager) StartAll()
  • func (m *Manager) StopAll()
• type RetryOnErrorOpt
• type ServerWorker
  • func NewServerWorker(name string, server *http.Server) *ServerWorker
  • func (w *ServerWorker) Start() error
  • func (w *ServerWorker) Stop() error
• type Worker

Variables

var (
	ErrServerIsAlreadyRun = errors.New("server is already run")
	ErrServerIsNotRun     = errors.New("server isn't run")
)

Functions

func SetLogger

func SetLogger(logger Logger)

Types

type CallbackFunc

type CallbackFunc func(ctx context.Context) error

type CallbackOpt

type CallbackOpt interface {
	// contains filtered or unexported methods
}

type CallbackWorker

type CallbackWorker struct {
	RetryOnError bool
	Retries      uint
	RetryTimeout time.Duration
	// contains filtered or unexported fields
}

func NewCallbackWorker

func NewCallbackWorker(name string, cb CallbackFunc, opts ...CallbackOpt) *CallbackWorker

func (*CallbackWorker) Start

func (w *CallbackWorker) Start() error

func (*CallbackWorker) Stop

func (w *CallbackWorker) Stop() error

type DummyLog

type DummyLog struct {
}

func (DummyLog) Error

func (DummyLog) Error(_ string, _ error, _ ...LogFields)

func (DummyLog) Info

func (DummyLog) Info(_ string, _ ...LogFields)

type LogFields

type LogFields map[string]interface{}

type Logger

type Logger interface {
	Info(msg string, fields ...LogFields)
	Error(msg string, err error, fields ...LogFields)
}

type Manager

type Manager struct {
	// contains filtered or unexported fields
}

func NewManager

func NewManager() *Manager

func NewManagerWithWorkers

func NewManagerWithWorkers(workers []Worker) *Manager

func (*Manager) AddWorker

func (m *Manager) AddWorker(worker Worker)

func (*Manager) AwaitAll

func (m *Manager) AwaitAll()

func (*Manager) IsRunning

func (m *Manager) IsRunning() bool

func (*Manager) StartAll

func (m *Manager) StartAll()

func (*Manager) StopAll

func (m *Manager) StopAll()

type RetryOnErrorOpt

type RetryOnErrorOpt struct {
	CallbackOpt

	Timeout     time.Duration
	MaxAttempts int
}

type ServerWorker

type ServerWorker struct {
	// contains filtered or unexported fields
}

func NewServerWorker

func NewServerWorker(name string, server *http.Server) *ServerWorker

func (*ServerWorker) Start

func (w *ServerWorker) Start() error

func (*ServerWorker) Stop

func (w *ServerWorker) Stop() error

type Worker

type Worker interface {
	// Start Run worker process in a current goroutine. An implementation
	// of this method mustn't return the execution context to the caller until finishing
	Start() error

	// Stop Graceful stop of the worker process. It have to finish all active
	// processes and return execution context to the caller.
	Stop() error
}