Generate event type identifiers

hack/codegen-events.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+
+# Dirty hack to generate boilerplate for event proto-messages
+
+import os
+import sys
+from subprocess import call
+
+if len(sys.argv) < 2:
+    print("usage: %s <proto-file>" % sys.argv[0])
+    exit(1)
+protoFile = sys.argv[1]
+
+eventNames = []
+with open(protoFile, "r") as fd:
+    for line in fd.readlines():
+        if not line.startswith("message"):
+            continue
+        eventNames.append(line.split(" ")[1])
+
+h, t = os.path.split(protoFile)
+outputFile = os.path.join(h, t.replace("proto", "gen.go"))
+with open(outputFile, "w") as fd:
+    fd.write("""// Code generated by hack/codegen-events.py. DO NOT EDIT.    
+package events
+
+import `github.com/golang/protobuf/proto`
+
+type EventType = string
+
+type Event interface {
+    proto.Message
+    Type() EventType
+} 
+
+""")
+
+    # Generate the constant declarations
+    consts = "const (\n"
+    for eventName in eventNames:
+        consts += "  %s EventType = \"%s\"\n" % ("Event" + eventName, eventName)
+    consts += ")\n\n"
+    fd.write(consts)
+
+    # Generate functions
+    for eventName in eventNames:
+        fd.write("""func (m *%s) Type() EventType {
+    return %s        
+}
+
+""" % (eventName, "Event" + eventName))
+
+
+call(["gofmt", "-w", outputFile])
+
+

pkg/api/events/events.proto

@@ -5,10 +5,6 @@ option go_package = "events";
 
 import "github.com/fission/fission-workflows/pkg/types/types.proto";
 
-message EventWrapper {
-    string any = 1;
-}
-
 //
 // Workflow
 //

pkg/controller/invocation/controller.go

@@ -140,13 +140,12 @@ func (cr *Controller) Notify(msg *fes.Notification) error {
 		"labels":       msg.Labels(),
 	}).Debugf("Controller event: %v", msg.EventType)
 
-	// TODO avoid struct creations
 	switch msg.EventType {
-	case events.TypeOf(&events.InvocationCompleted{}):
+	case events.EventInvocationCompleted:
 		fallthrough
-	case events.TypeOf(&events.InvocationCanceled{}):
+	case events.EventInvocationCanceled:
 		fallthrough
-	case events.TypeOf(&events.InvocationFailed{}):
+	case events.EventInvocationFailed:
 		wfi, ok := msg.Payload.(*aggregates.WorkflowInvocation)
 		if !ok {
 			log.Warn("Event did not contain invocation payload", msg)
@@ -155,11 +154,11 @@ func (cr *Controller) Notify(msg *fes.Notification) error {
 		cr.stateStore.Delete(wfi.ID())
 		cr.evalStore.Delete(wfi.ID())
 		log.Infof("Removed invocation %v from eval state", wfi.ID())
-	case events.TypeOf(&events.TaskFailed{}):
+	case events.EventTaskFailed:
 		fallthrough
-	case events.TypeOf(&events.TaskSucceeded{}):
+	case events.EventTaskSucceeded:
 		fallthrough
-	case events.TypeOf(&events.InvocationCreated{}):
+	case events.EventInvocationCreated:
 		wfi, ok := msg.Payload.(*aggregates.WorkflowInvocation)
 		if !ok {
 			panic(msg)

pkg/fes/util.go

@@ -2,13 +2,13 @@ package fes
 
 import (
 	"errors"
+	"reflect"
 	"strings"
 	"time"
 
 	"github.com/fission/fission-workflows/pkg/api/events"
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes"
-	"github.com/golang/protobuf/ptypes/any"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -59,21 +59,27 @@ type EventOpts struct {
 }
 
 func NewEvent(aggregate Aggregate, msg proto.Message) (*Event, error) {
-	var data *any.Any
 	if msg == nil {
-		return nil, errors.New("event cannot have no message")
+		return nil, errors.New("msg cannot have no message")
 	}
 
-	d, err := ptypes.MarshalAny(msg)
+	data, err := ptypes.MarshalAny(msg)
 	if err != nil {
 		return nil, err
 	}
-	data = d
+
+	var t string
+	if e, ok := msg.(events.Event); ok {
+		t = e.Type()
+	} else {
+		reflect.Indirect(reflect.ValueOf(msg)).Type().Name()
+	}
+
 	return &Event{
 		Aggregate: &aggregate,
 		Data:      data,
 		Timestamp: ptypes.TimestampNow(),
-		Type:      events.TypeOf(msg),
+		Type:      t,
 		Metadata:  map[string]string{},
 	}, nil
 }

pkg/fnenv/workflows/workflows.go

@@ -36,9 +36,9 @@ const (
 
 // TODO to fsm
 var terminationEvent = []string{
-	events.TypeOf(&events.InvocationCompleted{}),
-	events.TypeOf(&events.InvocationCanceled{}),
-	events.TypeOf(&events.InvocationFailed{}),
+	events.EventInvocationCompleted,
+	events.EventInvocationCanceled,
+	events.EventInvocationFailed,
 }
 
 // Runtime provides an abstraction of the workflow engine itself to use as a Task runtime environment.