Wrap builtin processors in middleware in examples (#1748)

* wrap builtin processors in middleware in examples

* simplify processor specification exporting

* disable schema middleware in error and filter processors

* generate processor specs

pkg/plugin/processor/builtin/impl/avro/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package avro
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }

pkg/plugin/processor/builtin/impl/base64/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package base64
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }

pkg/plugin/processor/builtin/impl/custom/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package custom
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }

pkg/plugin/processor/builtin/impl/error.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/Masterminds/sprig/v3"
 	"github.com/conduitio/conduit-commons/config"
+	"github.com/conduitio/conduit-commons/lang"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-processor-sdk"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
@@ -100,3 +101,17 @@ func (p *errorProcessor) errorMessage(record opencdc.Record) string {
 	}
 	return buf.String()
 }
+
+func (*errorProcessor) MiddlewareOptions() []sdk.ProcessorMiddlewareOption {
+	// disable schema middleware by default
+	return []sdk.ProcessorMiddlewareOption{
+		sdk.ProcessorWithSchemaEncodeConfig{
+			PayloadEnabled: lang.Ptr(false),
+			KeyEnabled:     lang.Ptr(false),
+		},
+		sdk.ProcessorWithSchemaDecodeConfig{
+			PayloadEnabled: lang.Ptr(false),
+			KeyEnabled:     lang.Ptr(false),
+		},
+	}
+}

pkg/plugin/processor/builtin/impl/field/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package field
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }

pkg/plugin/processor/builtin/impl/filter.go

@@ -18,6 +18,7 @@ import (
 	"context"
 
 	"github.com/conduitio/conduit-commons/config"
+	"github.com/conduitio/conduit-commons/lang"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-processor-sdk"
 	"github.com/conduitio/conduit/pkg/foundation/log"
@@ -54,3 +55,17 @@ func (p *filterProcessor) Process(_ context.Context, records []opencdc.Record) [
 	}
 	return out
 }
+
+func (*filterProcessor) MiddlewareOptions() []sdk.ProcessorMiddlewareOption {
+	// disable schema middleware by default
+	return []sdk.ProcessorMiddlewareOption{
+		sdk.ProcessorWithSchemaEncodeConfig{
+			PayloadEnabled: lang.Ptr(false),
+			KeyEnabled:     lang.Ptr(false),
+		},
+		sdk.ProcessorWithSchemaDecodeConfig{
+			PayloadEnabled: lang.Ptr(false),
+			KeyEnabled:     lang.Ptr(false),
+		},
+	}
+}

pkg/plugin/processor/builtin/impl/json/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package json
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }

pkg/plugin/processor/builtin/impl/unwrap/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package unwrap
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }

pkg/plugin/processor/builtin/impl/webhook/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package webhook
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }

pkg/plugin/processor/builtin/impl/z_examples_exporter_test.go

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build export_processors
+// This file is here to export processor specifications. It's named with a "z_"
+// prefix to ensure it's processed last by go generate.
+
+//go:generate env EXPORT_PROCESSORS=true go test -count=1 -tags export_processors .
 
 package impl
 
@@ -29,6 +32,8 @@ func TestMain(m *testing.M) {
 		os.Exit(code)
 	}
 
-	// tests passed, export the processors
-	exampleutil.ExportProcessors()
+	if os.Getenv("EXPORT_PROCESSORS") == "true" {
+		// tests passed and env var is included, export the processors
+		exampleutil.ExportProcessors()
+	}
 }