Merge pull request #17 from metrico/fluxpipe-library

Fluxpipelib

.github/workflows/go.yml

@@ -65,6 +65,7 @@ jobs:
       run: |
         strip fluxpipe-server
         strip fluxpipe-lambda
+        strip fluxpipelib.a
         upx fluxpipe-server
         
     - name: Get Timestamp
@@ -88,6 +89,7 @@ jobs:
         release_config: |
             fluxpipe-server
             fluxpipe-lambda
+            fluxpipelib.a
         tag_name: ${{ env.VERSION }}
         release_name: fluxpipe_${{ env.VERSION }}
         draft: false

build.sh

@@ -3,5 +3,10 @@ export PKG_CONFIG_PATH=$(pwd)
 
 echo "Building fluxpipe-server ..."
 go build -a -ldflags '-extldflags "-static -w -ldl"' -o fluxpipe-server fluxpipe-server.go
+
 echo "Building fluxpipe-lambda ..."
 go build -a -ldflags '-extldflags "-static -w -ldl"' -o fluxpipe-lambda fluxpipe-lambda.go
+
+echo "Building fluxpipe-lib ..."
+CGO_ENABLED=1 go build -buildmode=c-archive -o fluxpipelib.a fluxpipelib.go
+# CGO_ENABLED=1 go build -buildmode=c-shared -o fluxpipelib.dylib fluxpipelib.go

fluxpipelib.go

@@ -0,0 +1,111 @@
+package main
+
+import (
+	"C"
+	"bytes"
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/influxdata/flux"
+	"github.com/influxdata/flux/csv"
+	_ "github.com/influxdata/flux/fluxinit/static"
+	"github.com/influxdata/flux/lang"
+	"github.com/influxdata/flux/memory"
+	"github.com/influxdata/flux/runtime"
+
+	_fluxhttp "github.com/influxdata/flux/dependencies/http"
+	"github.com/influxdata/flux/dependencies/secret"
+	"github.com/influxdata/flux/dependencies/url"
+)
+
+
+// # CGO_ENABLED=1 go build -buildmode=c-shared -o fluxpipe.a fluxpipelib.go
+// # CGO_ENABLED=1 go build -buildmode=c-archive -o fluxpipe.a fluxpipelib.go
+
+var APPNAME = "fluxpipe-library"
+
+func runQuery(ctx context.Context, script string) (flux.Query, error) {
+
+	program, err := lang.Compile(ctx, script, runtime.Default, time.Now())
+	if err != nil {
+		return nil, err
+	}
+
+	q, err := program.Start(ctx, memory.DefaultAllocator)
+	if err != nil {
+		return nil, err
+	}
+	return q, nil
+}
+
+func NewCustomDependencies() flux.Deps {
+	validator := url.PassValidator{}
+	return flux.Deps{
+		Deps: flux.WrappedDeps{
+			HTTPClient: _fluxhttp.NewLimitedDefaultClient(validator),
+			// Default to having no filesystem, no secrets, and no url validation (always pass).
+			FilesystemService: nil,
+			SecretService:     secret.EnvironmentSecretService{},
+			URLValidator:      validator,
+		},
+	}
+}
+
+func exec(inputString string) (string, error) {
+
+	// CustomDeps produces a Custom set of dependencies including EnvironmentSecretService.
+	customValidator := url.PassValidator{}
+	customDeps := flux.Deps{
+		Deps: flux.WrappedDeps{
+			HTTPClient:        _fluxhttp.NewLimitedDefaultClient(customValidator),
+			FilesystemService: nil,
+			SecretService:     secret.EnvironmentSecretService{},
+			URLValidator:      customValidator,
+		},
+	}
+
+	// ctx := flux.NewDefaultDependencies().Inject(context.Background())
+	ctx := customDeps.Inject(context.Background())
+
+	q, err := runQuery(ctx, inputString)
+	if err != nil {
+		fmt.Println("unexpected error while creating query: %s", err)
+		return "", err
+	}
+
+	results := flux.NewResultIteratorFromQuery(q)
+	defer results.Release()
+
+	buf := bytes.NewBuffer(nil)
+	encoder := csv.NewMultiResultEncoder(csv.DefaultEncoderConfig())
+
+	if _, err := encoder.Encode(buf, results); err != nil {
+		return "", err
+	}
+
+	q.Done()
+
+	if q.Err() != nil {
+		fmt.Println("unexpected error from query execution: %s", q.Err())
+		return "", q.Err()
+
+	} else {
+		return buf.String(), nil
+	}
+}
+
+type FluxEvent struct {
+	Query string `json:"flux"`
+}
+
+//export Query
+func Query(query string) string {
+	res, err := exec(query)
+	if err != nil {
+		return fmt.Sprintf(`{"code":"invalid","message":"%v"}`, err.Error())
+	}
+	return res
+}
+
+func main() {}