diff --git a/requestoptions.go b/requestoptions.go index 962727562f..f6369ef2fc 100644 --- a/requestoptions.go +++ b/requestoptions.go @@ -205,6 +205,15 @@ func WithRequestBodyTimeout(timeout time.Duration) RequestOption { } } +// withWorker dispatches the request on a worker directly, no lookup +func withWorker(w *worker) RequestOption { + return func(o *frankenPHPContext) error { + o.worker = w + + return nil + } +} + // WithWorkerName sets the worker that should handle the request func WithWorkerName(name string) RequestOption { return func(o *frankenPHPContext) error { diff --git a/workerextension.go b/workerextension.go index 814eea1565..e873421c82 100644 --- a/workerextension.go +++ b/workerextension.go @@ -24,27 +24,45 @@ type extensionWorkers struct { internalWorker *worker } +// server is the one the worker belongs to, the fallback one for a global +// worker, nil before Init() +func (w *extensionWorkers) server() *Server { + switch { + case w.internalWorker == nil: + return nil + case w.internalWorker.server != nil: + return w.internalWorker.server + default: + return fallbackServer + } +} + // EXPERIMENTAL: SendRequest sends an HTTP request to the worker and writes the response to the provided ResponseWriter. func (w *extensionWorkers) SendRequest(rw http.ResponseWriter, r *http.Request) error { - fr, err := NewRequestWithContext( - r, - WithOriginalRequest(r), - WithWorkerName(w.name), - ) - - if err != nil { - return err + server := w.server() + if server == nil { + return ErrNotRunning } - return ServeHTTP(rw, fr) + // the worker itself, the name may be the empty one it was declared with + return server.ServeHTTP(rw, r, WithOriginalRequest(r), withWorker(w.internalWorker)) } func (w *extensionWorkers) NumThreads() int { + if w.internalWorker == nil { + return 0 + } + return w.internalWorker.countThreads() } // EXPERIMENTAL: SendMessage sends a message to the worker and waits for a response. func (w *extensionWorkers) SendMessage(ctx context.Context, message any, rw http.ResponseWriter) (any, error) { + // the worker stays referenced past Shutdown(), with no thread left to take the message + if server := w.server(); server == nil || !server.isRegistered.Load() { + return nil, ErrNotRunning + } + fc := newContextFromMessage(message, rw, ctx, w.internalWorker) err := w.internalWorker.handleRequest(fc) diff --git a/workerextension_test.go b/workerextension_test.go index e861ec97a3..e5e7bdde94 100644 --- a/workerextension_test.go +++ b/workerextension_test.go @@ -4,6 +4,7 @@ import ( "io" "net/http/httptest" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -77,6 +78,43 @@ func TestWorkerExtensionSendMessage(t *testing.T) { assert.Equal(t, "received message: Hello Workers", ret) } +// an extension worker without a name is still the one SendRequest() +// dispatches on +func TestWorkerExtensionSendRequestWithoutName(t *testing.T) { + t.Cleanup(Shutdown) + + externalWorker, o := WithExtensionWorkers("", "testdata/worker.php", 1) + require.NoError(t, Init(o)) + + w := httptest.NewRecorder() + require.NoError(t, externalWorker.SendRequest(w, httptest.NewRequest("GET", "http://example.com/index.php", nil))) + + body, err := io.ReadAll(w.Result().Body) + require.NoError(t, err) + assert.Contains(t, string(body), "Requests handled: 0") +} + +// SendMessage() after Shutdown() reports the stopped runtime instead of +// waiting for a thread +func TestWorkerExtensionSendMessageAfterShutdown(t *testing.T) { + externalWorker, o := WithExtensionWorkers("extensionWorkers", "testdata/message-worker.php", 1) + require.NoError(t, Init(o)) + Shutdown() + + errChan := make(chan error, 1) + go func() { + _, err := externalWorker.SendMessage(t.Context(), "Hello Workers", nil) + errChan <- err + }() + + select { + case err := <-errChan: + require.ErrorIs(t, err, ErrNotRunning) + case <-time.After(2 * time.Second): + t.Fatal("SendMessage() did not return after Shutdown()") + } +} + func TestErrorIf2WorkersHaveSameName(t *testing.T) { _, o1 := WithExtensionWorkers("duplicateWorker", "testdata/worker.php", 1) _, o2 := WithExtensionWorkers("duplicateWorker", "testdata/worker2.php", 1)