caffix/service

// Copyright © by Jeff Foley 2020-2023. All rights reserved.
// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
// SPDX-License-Identifier: Apache-2.0

package service

import (
    "errors"
    "sync"

    "go.uber.org/ratelimit"
)

// BaseService provides common mechanisms to all services implementing the Service interface.
type BaseService struct {
    sync.Mutex
    name   string
    runs   bool
    done   chan struct{}
    input  chan interface{}
    output chan interface{}
    rlock  sync.Mutex
    rlimit ratelimit.Limiter
    // The specific service embedding BaseService
    service Service
}

// NewBaseService returns an initialized BaseService object.
func NewBaseService(srv Service, name string) *BaseService {
    return &BaseService{
        name:    name,
        done:    make(chan struct{}),
        input:   make(chan interface{}),
        output:  make(chan interface{}, 10),
        service: srv,
    }
}

// Description implements the Service interface.
func (bas *BaseService) Description() string {
    return ""
}

// Start implements the Service interface.
func (bas *BaseService) Start() error {
    if bas.running() {
        return errors.New(bas.name + " has already been started")
    }

    bas.setRunning(true)
    return bas.service.OnStart()
}

// OnStart implements the Service interface.
func (bas *BaseService) OnStart() error {
    return nil
}

func (bas *BaseService) running() bool {
    bas.Lock()
    defer bas.Unlock()

    return bas.runs
}

func (bas *BaseService) setRunning(val bool) {
    bas.Lock()
    defer bas.Unlock()

    bas.runs = val
}

// Stop implements the Service interface.
func (bas *BaseService) Stop() error {
    if !bas.running() {
        return errors.New(bas.name + " is already stopped")
    }

    close(bas.done)
    finished := make(chan struct{})
    defer close(finished)

    drain := func(ch chan interface{}, finished chan struct{}) {
        for {
            select {
            case <-ch:
            case <-finished:
                return
            }
        }
    }

    go drain(bas.Input(), finished)
    go drain(bas.Output(), finished)

    defer bas.setRunning(false)
    return bas.service.OnStop()
}

// OnStop implements the Service interface.
func (bas *BaseService) OnStop() error {
    return nil
}

// Done implements the Service interface.
func (bas *BaseService) Done() <-chan struct{} {
    return bas.done
}

// Input implements the Service interface.
func (bas *BaseService) Input() chan interface{} {
    return bas.input
}

// HandlesReq implements the Service interface.
func (bas *BaseService) HandlesReq(req interface{}) bool {
    return true
}

// Output implements the Service interface.
func (bas *BaseService) Output() chan interface{} {
    return bas.output
}

// String implements the Stringer interface.
func (bas *BaseService) String() string {
    return bas.name
}

// SetRateLimit implements the Service interface.
func (bas *BaseService) SetRateLimit(persec int) {
    bas.rlock.Lock()
    defer bas.rlock.Unlock()

    if persec == 0 {
        bas.rlimit = nil
        return
    }
    bas.rlimit = ratelimit.New(persec, ratelimit.WithoutSlack)
}

// CheckRateLimit implements the Service interface.
func (bas *BaseService) CheckRateLimit() {
    bas.rlock.Lock()
    rlimit := bas.rlimit
    bas.rlock.Unlock()

    if rlimit != nil {
        rlimit.Take()
    }
}