examples: Add gRPC streaming example

examples/streaming/.gitignore

@@ -0,0 +1 @@
+myfile

examples/streaming/README.md

@@ -0,0 +1,38 @@
+# gRPC streaming Example
+
+This example builds a plugin & client which can stream larger amount of data
+between them while staying below reasonable message size limits of the gRPC
+protocol.
+
+## To execute
+
+Build the plugin
+
+```
+go build -o ./plugin/streamer ./plugin
+```
+
+launch client
+
+```
+go run main.go myfile
+```
+
+## To re-generate protobuf definitions
+
+Install protobuf tooling
+
+```
+brew install protobuf
+```
+```
+go install google.golang.org/protobuf/cmd/[email protected]
+go install google.golang.org/grpc/cmd/[email protected]
+```
+
+generate files
+
+```
+cd proto
+protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative streamer.proto
+```

examples/streaming/main.go

@@ -0,0 +1,87 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-plugin"
+	"github.com/hashicorp/go-plugin/examples/streaming/shared"
+	"google.golang.org/grpc"
+)
+
+func main() {
+	if len(os.Args) != 2 {
+		log.Fatal("expected path to file as an argument")
+	}
+	path := os.Args[1]
+
+	logger := hclog.New(&hclog.LoggerOptions{
+		Level:      hclog.Trace,
+		Output:     os.Stderr,
+		JSONFormat: true,
+	})
+
+	msgSizeLimit := 1000
+	chunkSize := 10
+
+	client := plugin.NewClient(&plugin.ClientConfig{
+		HandshakeConfig: plugin.HandshakeConfig{
+			ProtocolVersion:  1,
+			MagicCookieKey:   "BASIC_PLUGIN",
+			MagicCookieValue: "hello",
+		},
+		Plugins: map[string]plugin.Plugin{
+			"streamer": &shared.StreamerPlugin{},
+		},
+		Cmd: exec.Command("./plugin/streamer"),
+		AllowedProtocols: []plugin.Protocol{
+			plugin.ProtocolGRPC,
+		},
+		Logger: logger,
+		GRPCDialOptions: []grpc.DialOption{
+			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(msgSizeLimit)),
+			grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(msgSizeLimit)),
+		},
+	})
+	defer client.Kill()
+
+	logger.Debug("launching a client")
+
+	rpcClient, err := client.Client()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	raw, err := rpcClient.Dispense("streamer")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	ctx := context.Background()
+
+	streamer := raw.(shared.Streamer)
+	err = streamer.Configure(ctx, path, int64(chunkSize))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = streamer.Write(ctx, []byte("Lorem ipsum dolor sit amet"))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	logger.Debug("writing finished")
+
+	b, err := streamer.Read(ctx)
+	if err != nil {
+		log.Fatal(err)
+	}
+	logger.Debug(fmt.Sprintf("received %d bytes", len(b)), "bytes", string(b))
+}

examples/streaming/plugin/.gitignore

@@ -0,0 +1 @@
+streamer

examples/streaming/plugin/plugin.go

@@ -0,0 +1,82 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package main
+
+import (
+	"context"
+	"io"
+	"os"
+
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-plugin"
+	"github.com/hashicorp/go-plugin/examples/streaming/shared"
+)
+
+type FileStreamer struct {
+	logger hclog.Logger
+	path   string
+}
+
+func (fs *FileStreamer) Configure(ctx context.Context, path string, _ int64) error {
+	fs.path = path
+	return nil
+}
+
+func (fs *FileStreamer) Read(ctx context.Context) ([]byte, error) {
+	fs.logger.Debug("FileStreamer: Read", "path", fs.path)
+	f, err := os.OpenFile(fs.path, os.O_RDONLY, 0644)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	return io.ReadAll(f)
+}
+
+func (fs *FileStreamer) Write(ctx context.Context, b []byte) error {
+	fs.logger.Debug("FileStreamer: Write", "path", fs.path)
+	f, err := os.OpenFile(fs.path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	n, err := f.Write(b)
+	if err != nil {
+		return err
+	}
+	fs.logger.Debug("FileStreamer: Write finished", "bytes written", n)
+	return nil
+}
+
+var handshakeConfig = plugin.HandshakeConfig{
+	ProtocolVersion:  1,
+	MagicCookieKey:   "BASIC_PLUGIN",
+	MagicCookieValue: "hello",
+}
+
+func main() {
+	logger := hclog.New(&hclog.LoggerOptions{
+		Level:      hclog.Trace,
+		Output:     os.Stderr,
+		JSONFormat: true,
+	})
+
+	streamer := &FileStreamer{
+		logger: logger,
+	}
+	var pluginMap = map[string]plugin.Plugin{
+		"streamer": &shared.StreamerPlugin{
+			Impl: streamer,
+		},
+	}
+
+	logger.Debug("plugin launched, about to be served")
+
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: handshakeConfig,
+		Plugins:         pluginMap,
+		GRPCServer:      plugin.DefaultGRPCServer,
+		Logger:          logger,
+	})
+}