feat: Add basic implementation of callee callback transparency

cmd/dudirekta-example-closures-callee/main.go

@@ -7,7 +7,6 @@ import (
 	"net"
 	"time"
 
-	"github.com/pojntfx/dudirekta/pkg/closures"
 	"github.com/pojntfx/dudirekta/pkg/rpc"
 )
 
@@ -18,26 +17,18 @@ type local struct {
 func (s *local) Iterate(
 	ctx context.Context,
 	length int,
-	onIterationClosureID string,
+	onIteration func(ctx context.Context, i int) error,
 ) (int, error) {
-	peerID := rpc.GetRemoteID(ctx)
-
 	for i := 0; i < length; i++ {
-		for candidateIP, peer := range s.Peers() {
-			if candidateIP == peerID {
-				if _, err := peer.CallClosure(ctx, onIterationClosureID, []interface{}{i}); err != nil {
-					return -1, err
-				}
-			}
+		if err := onIteration(ctx, i); err != nil {
+			return -1, err
 		}
 	}
 
 	return length, nil
 }
 
-type remote struct {
-	CallClosure closures.CallClosureType
-}
+type remote struct{}
 
 func main() {
 	addr := flag.String("addr", ":1337", "Listen address")

cmd/dudirekta-example-closures-caller/main.go

@@ -18,11 +18,11 @@ type remote struct {
 	Iterate func(
 		ctx context.Context,
 		length int,
-		onIteration func(i int) error,
+		onIteration func(ctx context.Context, i int) error,
 	) (int, error)
 }
 
-func Iterate(callee remote, ctx context.Context, length int, onIteration func(i int) error) (int, error) {
+func Iterate(callee remote, ctx context.Context, length int, onIteration func(ctx context.Context, i int) error) (int, error) {
 	return callee.Iterate(ctx, length, onIteration)
 }
 
@@ -73,7 +73,7 @@ func main() {
 			for _, peer := range registry.Peers() {
 				switch line {
 				case "a\n":
-					length, err := Iterate(peer, ctx, 5, func(i int) error {
+					length, err := Iterate(peer, ctx, 5, func(ctx context.Context, i int) error {
 						log.Println("In iteration", i)
 
 						return nil
@@ -86,7 +86,7 @@ func main() {
 
 					log.Println(length)
 				case "b\n":
-					length, err := Iterate(peer, ctx, 10, func(i int) error {
+					length, err := Iterate(peer, ctx, 10, func(ctx context.Context, i int) error {
 						log.Println("In iteration", i)
 
 						return nil

pkg/closures/manager.go

@@ -10,13 +10,15 @@ import (
 )
 
 var (
-	errorType = reflect.TypeOf((*error)(nil)).Elem()
+	errorType   = reflect.TypeOf((*error)(nil)).Elem()
+	contextType = reflect.TypeOf((*context.Context)(nil)).Elem()
 
 	ErrInvalidReturn       = errors.New("invalid return, can only return an error or a value and an error")
 	ErrNotAFunction        = errors.New("not a function")
 	ErrInvalidArgsCount    = errors.New("invalid argument count")
 	ErrInvalidArg          = errors.New("invalid argument")
 	ErrClosureDoesNotExist = errors.New("closure does not exist")
+	ErrInvalidArgs         = errors.New("invalid arguments, first argument needs to be a context.Context")
 )
 
 type (
@@ -38,6 +40,14 @@ func createClosure(fn interface{}) (func(args ...interface{}) (interface{}, erro
 		return nil, ErrInvalidReturn
 	}
 
+	if functionType.NumIn() < 1 {
+		return nil, ErrInvalidArgs
+	}
+
+	if !functionType.In(0).Implements(contextType) {
+		return nil, ErrInvalidArgs
+	}
+
 	return func(args ...interface{}) (interface{}, error) {
 		if len(args) != functionType.NumIn() {
 			return nil, ErrInvalidArgsCount
@@ -49,7 +59,12 @@ func createClosure(fn interface{}) (func(args ...interface{}) (interface{}, erro
 				if argType.ConvertibleTo(functionType.In(i)) {
 					in[i] = reflect.ValueOf(arg).Convert(functionType.In(i))
 				} else {
-					return nil, ErrInvalidArg
+					// TODO: Drop this and the whole "there has to be a context in the callback function" requirementt
+					if functionType.In(i).Implements(contextType) {
+						in[i] = reflect.ValueOf(context.Background())
+					} else {
+						return nil, ErrInvalidArg
+					}
 				}
 			} else {
 				in[i] = reflect.ValueOf(arg)