refactor: extension worker

frankenphp.go

@@ -215,7 +215,7 @@ func Init(options ...Option) error {
 	registerExtensions()
 
 	// add registered external workers
-	for _, ew := range externalWorkers {
+	for _, ew := range extensionWorkers {
 		options = append(options, WithWorkers(ew.Name(), ew.FileName(), ew.GetMinThreads(), WithWorkerEnv(ew.Env())))
 	}
 

threadworker.go

@@ -20,12 +20,12 @@ type workerThread struct {
 	dummyContext    *frankenPHPContext
 	workerContext   *frankenPHPContext
 	backoff         *exponentialBackoff
-	externalWorker  WorkerExtension
+	externalWorker  Worker
 	isBootingScript bool // true if the worker has not reached frankenphp_handle_request yet
 }
 
 func convertToWorkerThread(thread *phpThread, worker *worker) {
-	externalWorker := externalWorkers[worker.name]
+	externalWorker := extensionWorkers[worker.name]
 
 	thread.setHandler(&workerThread{
 		state:  thread.state,

worker.go

@@ -55,7 +55,7 @@ func initWorkers(opt []workerOpt) error {
 				// create a pipe from the external worker to the main worker
 				// note: this is locked to the initial thread size the external worker requested
 				if workerThread, ok := thread.handler.(*workerThread); ok && workerThread.externalWorker != nil {
-					go startExternalWorkerPipe(w, workerThread.externalWorker, thread)
+					go startWorker(w, workerThread.externalWorker, thread)
 				}
 				workersReady.Done()
 			}()

threadFramework.go → workerextension.go

@@ -8,8 +8,8 @@ import (
 	"sync/atomic"
 )
 
-// EXPERIMENTAL: WorkerExtension allows you to register an external worker where instead of calling frankenphp handlers on
-// frankenphp_handle_request(), the ProvideRequest method is called. You are responsible for providing a standard
+// EXPERIMENTAL: Worker allows you to register a worker where instead of calling FrankenPHP handlers on
+// frankenphp_handle_request(), the ProvideRequest method is called. You may provide a standard
 // http.Request that will be conferred to the underlying worker script.
 //
 // A worker script with the provided Name and FileName will be registered, along with the provided
@@ -29,42 +29,43 @@ import (
 // Note: External workers receive the lowest priority when determining thread allocations. If GetMinThreads cannot be
 // allocated, then frankenphp will panic and provide this information to the user (who will need to allocate more
 // total threads). Don't be greedy.
-type WorkerExtension interface {
+type Worker interface {
 	Name() string
 	FileName() string
 	Env() PreparedEnv
 	GetMinThreads() int
 	ThreadActivatedNotification(threadId int)
 	ThreadDrainNotification(threadId int)
 	ThreadDeactivatedNotification(threadId int)
-	ProvideRequest() *WorkerRequest[any, any]
+	ProvideRequest() *WorkerRequest
+	InjectRequest(r *WorkerRequest)
 }
 
 // EXPERIMENTAL
-type WorkerRequest[P any, R any] struct {
+type WorkerRequest struct {
 	// The request for your worker script to handle
 	Request *http.Request
 	// Response is a response writer that provides the output of the provided request, it must not be nil to access the request body
 	Response http.ResponseWriter
 	// CallbackParameters is an optional field that will be converted in PHP types and passed as parameter to the PHP callback
-	CallbackParameters P
+	CallbackParameters any
 	// AfterFunc is an optional function that will be called after the request is processed with the original value, the return of the PHP callback, converted in Go types, is passed as parameter
-	AfterFunc func(callbackReturn R)
+	AfterFunc func(callbackReturn any)
 }
 
-var externalWorkers = make(map[string]WorkerExtension)
-var externalWorkerMutex sync.Mutex
+var extensionWorkers = make(map[string]Worker)
+var extensionWorkersMutex sync.Mutex
 
 // EXPERIMENTAL
-func RegisterExternalWorker(worker WorkerExtension) {
-	externalWorkerMutex.Lock()
-	defer externalWorkerMutex.Unlock()
+func RegisterWorker(worker Worker) {
+	extensionWorkersMutex.Lock()
+	defer extensionWorkersMutex.Unlock()
 
-	externalWorkers[worker.Name()] = worker
+	extensionWorkers[worker.Name()] = worker
 }
 
-// startExternalWorkerPipe creates a pipe from an external worker to the main worker.
-func startExternalWorkerPipe(w *worker, externalWorker WorkerExtension, thread *phpThread) {
+// startWorker creates a pipe from a worker to the main worker.
+func startWorker(w *worker, externalWorker Worker, thread *phpThread) {
 	for {
 		rq := externalWorker.ProvideRequest()
 
@@ -101,45 +102,61 @@ func startExternalWorkerPipe(w *worker, externalWorker WorkerExtension, thread *
 	}
 }
 
-type Worker struct {
-	ExtensionName  string
-	WorkerFileName string
-	WorkerEnv      PreparedEnv
-	MinThreads     int
-	RequestChan    chan *WorkerRequest[any, any]
-	ActivatedCount atomic.Int32
-	DrainCount     atomic.Int32
+func NewWorker(name, fileName string, minThreads int, env PreparedEnv) Worker {
+	return &defaultWorker{
+		name:           name,
+		fileName:       fileName,
+		env:            env,
+		minThreads:     minThreads,
+		requestChan:    make(chan *WorkerRequest),
+		activatedCount: atomic.Int32{},
+		drainCount:     atomic.Int32{},
+	}
+}
+
+type defaultWorker struct {
+	name           string
+	fileName       string
+	env            PreparedEnv
+	minThreads     int
+	requestChan    chan *WorkerRequest
+	activatedCount atomic.Int32
+	drainCount     atomic.Int32
+}
+
+func (w *defaultWorker) Name() string {
+	return w.name
 }
 
-func (w *Worker) Name() string {
-	return w.ExtensionName
+func (w *defaultWorker) FileName() string {
+	return w.fileName
 }
 
-func (w *Worker) FileName() string {
-	return w.WorkerFileName
+func (w *defaultWorker) Env() PreparedEnv {
+	return w.env
 }
 
-func (w *Worker) Env() PreparedEnv {
-	return w.WorkerEnv
+func (w *defaultWorker) GetMinThreads() int {
+	return w.minThreads
 }
 
-func (w *Worker) GetMinThreads() int {
-	return w.MinThreads
+func (w *defaultWorker) ThreadActivatedNotification(_ int) {
+	w.activatedCount.Add(1)
 }
 
-func (w *Worker) ThreadActivatedNotification(threadId int) {
-	w.ActivatedCount.Add(1)
+func (w *defaultWorker) ThreadDrainNotification(_ int) {
+	w.drainCount.Add(1)
 }
 
-func (w *Worker) ThreadDrainNotification(threadId int) {
-	w.DrainCount.Add(1)
+func (w *defaultWorker) ThreadDeactivatedNotification(_ int) {
+	w.drainCount.Add(-1)
+	w.activatedCount.Add(-1)
 }
 
-func (w *Worker) ThreadDeactivatedNotification(threadId int) {
-	w.DrainCount.Add(-1)
-	w.ActivatedCount.Add(-1)
+func (w *defaultWorker) ProvideRequest() *WorkerRequest {
+	return <-w.requestChan
 }
 
-func (w *Worker) ProvideRequest() *WorkerRequest[any, any] {
-	return <-w.RequestChan
+func (w *defaultWorker) InjectRequest(r *WorkerRequest) {
+	w.requestChan <- r
 }

threadFramework_test.go → workerextension_test.go

@@ -3,51 +3,30 @@ package frankenphp
 import (
 	"io"
 	"net/http/httptest"
-	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-// mockWorkerExtension implements the WorkerExtension interface
-type mockWorkerExtension struct {
+// mockWorker implements the Worker interface
+type mockWorker struct {
 	Worker
 }
 
-func newMockWorkerExtension(name, fileName string, minThreads int) *mockWorkerExtension {
-	return &mockWorkerExtension{
-		Worker: Worker{
-			ExtensionName:  name,
-			WorkerFileName: fileName,
-			WorkerEnv:      nil,
-			MinThreads:     minThreads,
-			RequestChan:    make(chan *WorkerRequest[any, any], minThreads),
-			ActivatedCount: atomic.Int32{},
-			DrainCount:     atomic.Int32{},
-		},
-	}
-}
-
-func (m *mockWorkerExtension) InjectRequest(r *WorkerRequest[any, any]) {
-	m.RequestChan <- r
-}
-
-func (m *mockWorkerExtension) GetActivatedCount() int {
-	return int(m.ActivatedCount.Load())
-}
-
 func TestWorkerExtension(t *testing.T) {
-	// Create a mock extension
-	mockExt := newMockWorkerExtension("mockWorker", "testdata/worker.php", 1)
+	// Create a mock worker extension
+	mockExt := &mockWorker{
+		Worker: NewWorker("mockWorker", "testdata/worker.php", 1, nil),
+	}
 
 	// Register the mock extension
-	RegisterExternalWorker(mockExt)
+	RegisterWorker(mockExt)
 
 	// Clean up external workers after test to avoid interfering with other tests
 	defer func() {
-		delete(externalWorkers, mockExt.Name())
+		delete(extensionWorkers, mockExt.Name())
 	}()
 
 	// Initialize FrankenPHP with a worker that has a different name than our extension
@@ -59,10 +38,10 @@ func TestWorkerExtension(t *testing.T) {
 	time.Sleep(100 * time.Millisecond)
 
 	// Verify that the extension's thread was activated
-	assert.GreaterOrEqual(t, mockExt.GetActivatedCount(), 1, "Thread should have been activated")
+	assert.GreaterOrEqual(t, int(mockExt.Worker.(*defaultWorker).activatedCount.Load()), 1, "Thread should have been activated")
 
 	// Create a test request
-	req := httptest.NewRequest("GET", "http://example.com/test/?foo=bar", nil)
+	req := httptest.NewRequest("GET", "https://example.com/test/?foo=bar", nil)
 	req.Header.Set("X-Test-Header", "test-value")
 
 	w := httptest.NewRecorder()
@@ -71,7 +50,7 @@ func TestWorkerExtension(t *testing.T) {
 	done := make(chan struct{})
 
 	// Inject the request into the worker through the extension
-	mockExt.InjectRequest(&WorkerRequest[any, any]{
+	mockExt.InjectRequest(&WorkerRequest{
 		Request:  req,
 		Response: w,
 		AfterFunc: func(callbackReturn any) {