fix: workflow dispatcher uses internal handler instead of raw gRPC proxy
CI / CI (pull_request) Successful in 3m19s
CI / CI (pull_request) Successful in 3m19s
- Wire WorkflowAdapter as internal handler (JSON-to-gRPC bridge) - Add ServeHTTP to WorkflowAdapter: maps X-Resource to Temporal action - Rename resource 'start' to 'execute' for consistency - Add GET methods for describe/list/history resources - Remove unused workflowAdapterImpl variable - SDK clients can now send JSON, gateway translates to gRPC
This commit is contained in:
+3
-2
@@ -76,9 +76,11 @@ func main() {
|
|||||||
|
|
||||||
// Add workflow service adapter (uses Temporal handler for gRPC forwarding)
|
// Add workflow service adapter (uses Temporal handler for gRPC forwarding)
|
||||||
workflowSpec := serviceadapter.GetWorkflowSpec()
|
workflowSpec := serviceadapter.GetWorkflowSpec()
|
||||||
|
workflowAdapterHandler := serviceadapter.NewWorkflowAdapter(temporalHandler)
|
||||||
workflowAdapter := &serviceadapter.ServiceAdapter{
|
workflowAdapter := &serviceadapter.ServiceAdapter{
|
||||||
Namespace: "temporal",
|
Namespace: "temporal",
|
||||||
ServiceName: "workflow",
|
ServiceName: "workflow",
|
||||||
|
Handler: workflowAdapterHandler,
|
||||||
Spec: *workflowSpec,
|
Spec: *workflowSpec,
|
||||||
}
|
}
|
||||||
_ = registry.Add(workflowAdapter)
|
_ = registry.Add(workflowAdapter)
|
||||||
@@ -120,8 +122,7 @@ func main() {
|
|||||||
dispatcher := serviceadapter.NewDispatcher(registry, jwtValidator)
|
dispatcher := serviceadapter.NewDispatcher(registry, jwtValidator)
|
||||||
|
|
||||||
// Wire workflow adapter to temporal handler for proper request forwarding
|
// Wire workflow adapter to temporal handler for proper request forwarding
|
||||||
workflowAdapterImpl := serviceadapter.NewWorkflowAdapter(temporalHandler)
|
// workflowAdapterHandler (above) handles JSON-to-gRPC translation for workflow service
|
||||||
_ = workflowAdapterImpl // The dispatcher will call temporal handler directly for gRPC
|
|
||||||
|
|
||||||
// Create router that handles health endpoints, X-Service (ServiceAdapter) routing,
|
// Create router that handles health endpoints, X-Service (ServiceAdapter) routing,
|
||||||
// temporal endpoints, and passes others to upstream handler
|
// temporal endpoints, and passes others to upstream handler
|
||||||
|
|||||||
@@ -84,6 +84,68 @@ func (wa *WorkflowAdapter) HandleUpdate(w http.ResponseWriter, r *http.Request)
|
|||||||
wa.forwardToTemporal(w, r)
|
wa.forwardToTemporal(w, r)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// resourceToAction maps X-Resource names to Temporal action names.
|
||||||
|
var resourceToAction = map[string]string{
|
||||||
|
"execute": "START_WORKFLOW",
|
||||||
|
"describe": "DESCRIBE_WORKFLOW",
|
||||||
|
"list": "LIST_WORKFLOWS",
|
||||||
|
"history": "GET_WORKFLOW_HISTORY",
|
||||||
|
"terminate": "TERMINATE_WORKFLOW",
|
||||||
|
"cancel": "CANCEL_WORKFLOW",
|
||||||
|
"signal": "SIGNAL_WORKFLOW",
|
||||||
|
"query": "QUERY_WORKFLOW",
|
||||||
|
"reset": "RESET_WORKFLOW",
|
||||||
|
"update": "UPDATE_WORKFLOW",
|
||||||
|
}
|
||||||
|
|
||||||
|
// ServeHTTP implements http.Handler for X-Service: workflow routing.
|
||||||
|
// Maps X-Resource header to Temporal action, injects action into body,
|
||||||
|
// and forwards to the temporal handler.
|
||||||
|
func (wa *WorkflowAdapter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||||
|
resource := r.Header.Get("X-Resource")
|
||||||
|
action, ok := resourceToAction[resource]
|
||||||
|
if !ok {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusNotFound)
|
||||||
|
fmt.Fprintf(w, `{"error":"unknown workflow resource: %s"}`, resource)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read body, inject action, forward
|
||||||
|
body, err := io.ReadAll(r.Body)
|
||||||
|
if err != nil {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusBadRequest)
|
||||||
|
fmt.Fprintf(w, `{"error":"failed to read body: %s"}`, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var payload map[string]interface{}
|
||||||
|
if len(body) > 0 {
|
||||||
|
if err := json.Unmarshal(body, &payload); err != nil {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusBadRequest)
|
||||||
|
fmt.Fprintf(w, `{"error":"invalid JSON: %s"}`, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
payload = make(map[string]interface{})
|
||||||
|
}
|
||||||
|
|
||||||
|
// Inject action into body for temporal handler
|
||||||
|
payload["action"] = action
|
||||||
|
if _, ok := payload["namespace"]; !ok {
|
||||||
|
payload["namespace"] = "default"
|
||||||
|
}
|
||||||
|
|
||||||
|
newBody, _ := json.Marshal(payload)
|
||||||
|
r.Body = io.NopCloser(bytes.NewReader(newBody))
|
||||||
|
r.ContentLength = int64(len(newBody))
|
||||||
|
r.URL.Path = "/workflow"
|
||||||
|
|
||||||
|
wa.temporalHandler.ServeHTTP(w, r)
|
||||||
|
}
|
||||||
|
|
||||||
// forwardToTemporal reads the request body, ensures namespace is specified,
|
// forwardToTemporal reads the request body, ensures namespace is specified,
|
||||||
// and forwards to the temporal handler.
|
// and forwards to the temporal handler.
|
||||||
func (wa *WorkflowAdapter) forwardToTemporal(w http.ResponseWriter, r *http.Request) {
|
func (wa *WorkflowAdapter) forwardToTemporal(w http.ResponseWriter, r *http.Request) {
|
||||||
@@ -134,7 +196,7 @@ func GetWorkflowSpec() *Spec {
|
|||||||
Retryable: true,
|
Retryable: true,
|
||||||
Resources: []Resource{
|
Resources: []Resource{
|
||||||
{
|
{
|
||||||
Name: "start",
|
Name: "execute",
|
||||||
Methods: []Method{
|
Methods: []Method{
|
||||||
{
|
{
|
||||||
Verb: "POST",
|
Verb: "POST",
|
||||||
@@ -151,6 +213,9 @@ func GetWorkflowSpec() *Spec {
|
|||||||
{
|
{
|
||||||
Name: "describe",
|
Name: "describe",
|
||||||
Methods: []Method{
|
Methods: []Method{
|
||||||
|
{
|
||||||
|
Verb: "GET",
|
||||||
|
},
|
||||||
{
|
{
|
||||||
Verb: "POST",
|
Verb: "POST",
|
||||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/DescribeWorkflowExecution",
|
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/DescribeWorkflowExecution",
|
||||||
@@ -166,6 +231,9 @@ func GetWorkflowSpec() *Spec {
|
|||||||
{
|
{
|
||||||
Name: "list",
|
Name: "list",
|
||||||
Methods: []Method{
|
Methods: []Method{
|
||||||
|
{
|
||||||
|
Verb: "GET",
|
||||||
|
},
|
||||||
{
|
{
|
||||||
Verb: "POST",
|
Verb: "POST",
|
||||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/ListWorkflowExecutions",
|
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/ListWorkflowExecutions",
|
||||||
@@ -181,6 +249,9 @@ func GetWorkflowSpec() *Spec {
|
|||||||
{
|
{
|
||||||
Name: "history",
|
Name: "history",
|
||||||
Methods: []Method{
|
Methods: []Method{
|
||||||
|
{
|
||||||
|
Verb: "GET",
|
||||||
|
},
|
||||||
{
|
{
|
||||||
Verb: "POST",
|
Verb: "POST",
|
||||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/GetWorkflowExecutionHistory",
|
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/GetWorkflowExecutionHistory",
|
||||||
|
|||||||
Reference in New Issue
Block a user