Coroutines

.gitignore

@@ -0,0 +1 @@
+*_durable.go

Makefile

@@ -1,4 +1,4 @@
-.PHONY: fmt lint test
+.PHONY: clean coroc fmt lint test integration-test clean coroc
 
 fmt:
 	go fmt ./...
@@ -8,3 +8,16 @@ lint:
 
 test:
 	go test ./...
+
+integration-test: clean coroc
+	go run ./dispatchtest/integration # volatile mode
+	coroc ./dispatchtest/integration
+	go run -tags durable ./dispatchtest/integration # durable mode
+
+clean:
+	find . -name '*_durable.go' -delete
+
+coroc:
+	@which coroc &>/dev/null \
+		|| echo "Installing coroc..." \
+		&& go install github.com/dispatchrun/coroutine/compiler/cmd/coroc@latest

client.go

@@ -1,17 +1,29 @@
+//go:build !durable
+
 package dispatch
 
 import (
 	"context"
 	"fmt"
 	"net/http"
 	"os"
+	_ "unsafe"
 
 	"buf.build/gen/go/stealthrocket/dispatch-proto/connectrpc/go/dispatch/sdk/v1/sdkv1connect"
 	sdkv1 "buf.build/gen/go/stealthrocket/dispatch-proto/protocolbuffers/go/dispatch/sdk/v1"
 	"connectrpc.com/connect"
 	"connectrpc.com/validate"
+	"github.com/dispatchrun/dispatch-go/dispatchproto"
 )
 
+const defaultApiUrl = "https://api.dispatch.run"
+
+// Call is a function call.
+type Call = dispatchproto.Call
+
+// ID is an identifier for a dispatched function call.
+type ID = dispatchproto.ID
+
 // Client is a client for the Dispatch API.
 //
 // The Client can be used to dispatch function calls.
@@ -21,14 +33,16 @@ type Client struct {
 	apiUrl        string
 	env           []string
 	httpClient    *http.Client
+	opts          []ClientOption
 
 	client sdkv1connect.DispatchServiceClient
 }
 
 // NewClient creates a Client.
 func NewClient(opts ...ClientOption) (*Client, error) {
 	c := &Client{
-		env: os.Environ(),
+		env:  os.Environ(),
+		opts: opts,
 	}
 	for _, opt := range opts {
 		opt.configureClient(c)
@@ -46,7 +60,7 @@ func NewClient(opts ...ClientOption) (*Client, error) {
 		c.apiUrl = getenv(c.env, "DISPATCH_API_URL")
 	}
 	if c.apiUrl == "" {
-		c.apiUrl = DefaultApiUrl
+		c.apiUrl = defaultApiUrl
 	}
 
 	if c.httpClient == nil {
@@ -94,14 +108,12 @@ func APIKey(apiKey string) ClientOption {
 // APIUrl sets the URL of the Dispatch API.
 //
 // It defaults to the value of the DISPATCH_API_URL environment variable,
-// or DefaultApiUrl if DISPATCH_API_URL is unset.
+// or the default API URL (https://api.dispatch.run) if DISPATCH_API_URL
+// is unset.
 func APIUrl(apiUrl string) ClientOption {
 	return clientOptionFunc(func(c *Client) { c.apiUrl = apiUrl })
 }
 
-// DefaultApiUrl is the default Dispatch API URL.
-const DefaultApiUrl = "https://api.dispatch.run"
-
 // Dispatch dispatches a function call.
 func (c *Client) Dispatch(ctx context.Context, call Call) (ID, error) {
 	batch := c.Batch()
@@ -138,10 +150,13 @@ func (b *Batch) Reset() {
 // Add adds calls to the batch.
 func (b *Batch) Add(calls ...Call) {
 	for i := range calls {
-		b.calls = append(b.calls, calls[i].proto)
+		b.calls = append(b.calls, callProto(calls[i]))
 	}
 }
 
+//go:linkname callProto github.com/dispatchrun/dispatch-go/dispatchproto.callProto
+func callProto(r dispatchproto.Call) *sdkv1.Call
+
 // Dispatch dispatches the batch of function calls.
 func (b *Batch) Dispatch(ctx context.Context) ([]ID, error) {
 	req := connect.NewRequest(&sdkv1.DispatchRequest{Calls: b.calls})

client_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/dispatchrun/dispatch-go"
+	"github.com/dispatchrun/dispatch-go/dispatchproto"
 	"github.com/dispatchrun/dispatch-go/dispatchtest"
 )
 
@@ -18,7 +19,7 @@ func TestClient(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	call := dispatch.NewCall("http://example.com", "function1", dispatch.Input(dispatch.Int(11)))
+	call := dispatchproto.NewCall("http://example.com", "function1", dispatchproto.Int(11))
 
 	_, err = client.Dispatch(context.Background(), call)
 	if err != nil {
@@ -27,7 +28,7 @@ func TestClient(t *testing.T) {
 
 	recorder.Assert(t, dispatchtest.DispatchRequest{
 		Header: http.Header{"Authorization": []string{"Bearer foobar"}},
-		Calls:  []dispatch.Call{call},
+		Calls:  []dispatchproto.Call{call},
 	})
 }
 
@@ -43,7 +44,7 @@ func TestClientEnvConfig(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	call := dispatch.NewCall("http://example.com", "function1", dispatch.Input(dispatch.Int(11)))
+	call := dispatchproto.NewCall("http://example.com", "function1", dispatchproto.Int(11))
 
 	_, err = client.Dispatch(context.Background(), call)
 	if err != nil {
@@ -52,7 +53,7 @@ func TestClientEnvConfig(t *testing.T) {
 
 	recorder.Assert(t, dispatchtest.DispatchRequest{
 		Header: http.Header{"Authorization": []string{"Bearer foobar"}},
-		Calls:  []dispatch.Call{call},
+		Calls:  []dispatchproto.Call{call},
 	})
 }
 
@@ -65,10 +66,10 @@ func TestClientBatch(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	call1 := dispatch.NewCall("http://example.com", "function1", dispatch.Input(dispatch.Int(11)))
-	call2 := dispatch.NewCall("http://example.com", "function2", dispatch.Input(dispatch.Int(22)))
-	call3 := dispatch.NewCall("http://example.com", "function3", dispatch.Input(dispatch.Int(33)))
-	call4 := dispatch.NewCall("http://example2.com", "function4", dispatch.Input(dispatch.Int(44)))
+	call1 := dispatchproto.NewCall("http://example.com", "function1", dispatchproto.Int(11))
+	call2 := dispatchproto.NewCall("http://example.com", "function2", dispatchproto.Int(22))
+	call3 := dispatchproto.NewCall("http://example.com", "function3", dispatchproto.Int(33))
+	call4 := dispatchproto.NewCall("http://example2.com", "function4", dispatchproto.Int(44))
 
 	batch := client.Batch()
 	batch.Add(call1, call2)
@@ -88,11 +89,11 @@ func TestClientBatch(t *testing.T) {
 	recorder.Assert(t,
 		dispatchtest.DispatchRequest{
 			Header: http.Header{"Authorization": []string{"Bearer foobar"}},
-			Calls:  []dispatch.Call{call1, call2},
+			Calls:  []dispatchproto.Call{call1, call2},
 		},
 		dispatchtest.DispatchRequest{
 			Header: http.Header{"Authorization": []string{"Bearer foobar"}},
-			Calls:  []dispatch.Call{call3, call4},
+			Calls:  []dispatchproto.Call{call3, call4},
 		})
 }