[plc4x] 01/02: refactor(plc4go/spi): use Executor interface

[sr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git

commit be227a7ad0989cad0e29e56fd197472477457272
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Mar 21 13:06:44 2023 +0100

    refactor(plc4go/spi): use Executor interface
---
 plc4go/spi/RequestTransactionManager.go |  8 ++++----
 plc4go/spi/utils/WorkerPool.go          | 29 ++++++++++++++++++-----------
 2 files changed, 22 insertions(+), 15 deletions(-)

diff --git a/plc4go/spi/RequestTransactionManager.go b/plc4go/spi/RequestTransactionManager.go
index b1d4b78716..aca4dc4d41 100644
--- a/plc4go/spi/RequestTransactionManager.go
+++ b/plc4go/spi/RequestTransactionManager.go
@@ -38,7 +38,7 @@ import (
 var sharedExecutorInstance utils.Executor // shared instance
 
 func init() {
-	sharedExecutorInstance = *utils.NewFixedSizeExecutor(runtime.NumCPU(), 100, utils.WithExecutorOptionTracerWorkers(config.TraceTransactionManagerWorkers))
+	sharedExecutorInstance = utils.NewFixedSizeExecutor(runtime.NumCPU(), 100, utils.WithExecutorOptionTracerWorkers(config.TraceTransactionManagerWorkers))
 	sharedExecutorInstance.Start()
 }
 
@@ -68,7 +68,7 @@ type RequestTransactionManager struct {
 	// Important, this is a FIFO Queue for Fairness!
 	workLog      list.List
 	workLogMutex sync.RWMutex
-	executor     *utils.Executor
+	executor     utils.Executor
 }
 
 // NewRequestTransactionManager creates a new RequestTransactionManager
@@ -77,7 +77,7 @@ func NewRequestTransactionManager(numberOfConcurrentRequests int, requestTransac
 		numberOfConcurrentRequests: numberOfConcurrentRequests,
 		transactionId:              0,
 		workLog:                    *list.New(),
-		executor:                   &sharedExecutorInstance,
+		executor:                   sharedExecutorInstance,
 	}
 	for _, requestTransactionManagerOption := range requestTransactionManagerOptions {
 		requestTransactionManagerOption(requestTransactionManager)
@@ -88,7 +88,7 @@ func NewRequestTransactionManager(numberOfConcurrentRequests int, requestTransac
 type RequestTransactionManagerOption func(requestTransactionManager *RequestTransactionManager)
 
 // WithCustomExecutor sets a custom Executor for the RequestTransactionManager
-func WithCustomExecutor(executor *utils.Executor) RequestTransactionManagerOption {
+func WithCustomExecutor(executor utils.Executor) RequestTransactionManagerOption {
 	return func(requestTransactionManager *RequestTransactionManager) {
 		requestTransactionManager.executor = executor
 	}
diff --git a/plc4go/spi/utils/WorkerPool.go b/plc4go/spi/utils/WorkerPool.go
index 7a6de6b087..379a2c3bd7 100644
--- a/plc4go/spi/utils/WorkerPool.go
+++ b/plc4go/spi/utils/WorkerPool.go
@@ -36,7 +36,7 @@ type Worker struct {
 	shutdown    atomic.Bool
 	runnable    Runnable
 	interrupted atomic.Bool
-	executor    *Executor
+	executor    *executor
 }
 
 func (w *Worker) work() {
@@ -85,7 +85,14 @@ func (w *WorkItem) String() string {
 	return fmt.Sprintf("Workitem{wid:%d}", w.workItemId)
 }
 
-type Executor struct {
+type Executor interface {
+	Start()
+	Stop()
+	Submit(ctx context.Context, workItemId int32, runnable Runnable) CompletionFuture
+	IsRunning() bool
+}
+
+type executor struct {
 	running      bool
 	shutdown     bool
 	stateChange  sync.Mutex
@@ -94,14 +101,14 @@ type Executor struct {
 	traceWorkers bool
 }
 
-func NewFixedSizeExecutor(numberOfWorkers, queueDepth int, options ...ExecutorOption) *Executor {
+func NewFixedSizeExecutor(numberOfWorkers, queueDepth int, options ...ExecutorOption) Executor {
 	workers := make([]*Worker, numberOfWorkers)
 	for i := 0; i < numberOfWorkers; i++ {
 		workers[i] = &Worker{
 			id: i,
 		}
 	}
-	executor := &Executor{
+	executor := &executor{
 		queue:  make(chan WorkItem, queueDepth),
 		worker: workers,
 	}
@@ -114,15 +121,15 @@ func NewFixedSizeExecutor(numberOfWorkers, queueDepth int, options ...ExecutorOp
 	return executor
 }
 
-type ExecutorOption func(*Executor)
+type ExecutorOption func(*executor)
 
 func WithExecutorOptionTracerWorkers(traceWorkers bool) ExecutorOption {
-	return func(executor *Executor) {
+	return func(executor *executor) {
 		executor.traceWorkers = traceWorkers
 	}
 }
 
-func (e *Executor) Submit(ctx context.Context, workItemId int32, runnable Runnable) CompletionFuture {
+func (e *executor) Submit(ctx context.Context, workItemId int32, runnable Runnable) CompletionFuture {
 	log.Trace().Int32("workItemId", workItemId).Msg("Submitting runnable")
 	completionFuture := &future{}
 	select {
@@ -140,7 +147,7 @@ func (e *Executor) Submit(ctx context.Context, workItemId int32, runnable Runnab
 	return completionFuture
 }
 
-func (e *Executor) Start() {
+func (e *executor) Start() {
 	e.stateChange.Lock()
 	defer e.stateChange.Unlock()
 	if e.running || e.shutdown {
@@ -154,10 +161,10 @@ func (e *Executor) Start() {
 	}
 }
 
-func (e *Executor) Stop() {
+func (e *executor) Stop() {
 	e.stateChange.Lock()
 	defer e.stateChange.Unlock()
-	if !e.running {
+	if !e.running || e.shutdown {
 		return
 	}
 	e.shutdown = true
@@ -171,7 +178,7 @@ func (e *Executor) Stop() {
 	e.shutdown = false
 }
 
-func (e *Executor) IsRunning() bool {
+func (e *executor) IsRunning() bool {
 	return e.running && !e.shutdown
 }