Tidy up json_documents scanner

Author: Jeffail

CHANGELOG.md

@@ -17,6 +17,7 @@ All notable changes to this project will be documented in this file.
 - Field `targets_input` added to the `azure_blob_storage` input.
 - New `reject_errored` output.
 - New `nats_request_reply` processor.
+- New `json_documents` scanner.
 
 ### Fixed
 

internal/impl/pure/scanner_json.go

@@ -2,55 +2,37 @@ package pure
 
 import (
 	"context"
-	json "encoding/json"
-	"errors"
+	"encoding/json"
 	"io"
 
 	"github.com/benthosdev/benthos/v4/public/service"
 )
 
-const (
-	sjsonFieldContinueOnError = "continue_on_error"
-)
-
 func jsonDocumentScannerSpec() *service.ConfigSpec {
 	return service.NewConfigSpec().
 		Stable().
+		Version("4.27.0").
 		Summary("Consumes a stream of one or more JSON documents.").
-		Fields(
-			service.NewBoolField(sjsonFieldContinueOnError).
-				Description("If a JSON decoding fails due to any error emit an empty message marked with the error and then continue consuming subsequent documents when possible.").
-				Default(false),
-		)
+		// Just a placeholder empty object as we don't have any fields yet
+		Field(service.NewObjectField("").Default(map[string]any{}))
 }
 
 func init() {
 	err := service.RegisterBatchScannerCreator("json_documents", jsonDocumentScannerSpec(),
 		func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchScannerCreator, error) {
-			return jsonDocumentScannerFromParsed(conf)
+			return &jsonDocumentScannerCreator{}, nil
 		})
 	if err != nil {
 		panic(err)
 	}
 }
 
-func jsonDocumentScannerFromParsed(conf *service.ParsedConfig) (l *jsonDocumentScannerCreator, err error) {
-	l = &jsonDocumentScannerCreator{}
-	if l.continueOnError, err = conf.FieldBool(sjsonFieldContinueOnError); err != nil {
-		return
-	}
-	return
-}
-
-type jsonDocumentScannerCreator struct {
-	continueOnError bool
-}
+type jsonDocumentScannerCreator struct{}
 
 func (js *jsonDocumentScannerCreator) Create(rdr io.ReadCloser, aFn service.AckFunc, details *service.ScannerSourceDetails) (service.BatchScanner, error) {
 	return service.AutoAggregateBatchScannerAcks(&jsonDocumentScanner{
-		d:               json.NewDecoder(rdr),
-		r:               rdr,
-		continueOnError: js.continueOnError,
+		d: json.NewDecoder(rdr),
+		r: rdr,
 	}, aFn), nil
 }
 
@@ -61,27 +43,21 @@ func (js *jsonDocumentScannerCreator) Close(context.Context) error {
 type jsonDocumentScanner struct {
 	d *json.Decoder
 	r io.ReadCloser
-
-	continueOnError bool
 }
 
 func (js *jsonDocumentScanner) NextBatch(ctx context.Context) (service.MessageBatch, error) {
 	if js.r == nil {
 		return nil, io.EOF
 	}
 
-	msg := service.NewMessage(nil)
-
 	var jsonDocObj any
-
 	if err := js.d.Decode(&jsonDocObj); err != nil {
-		msg.SetError(err)
-		if errors.Is(err, io.EOF) || !js.continueOnError {
-			return nil, err
-		}
+		_ = js.r.Close()
+		js.r = nil
+		return nil, err
 	}
 
-	// Decode will determine whether a the jsonObj is of type map[string]interface{} or []interface{}
+	msg := service.NewMessage(nil)
 	msg.SetStructuredMut(jsonDocObj)
 
 	return service.MessageBatch{msg}, nil

internal/impl/pure/scanner_json_test.go

@@ -1,8 +1,12 @@
 package pure_test
 
 import (
+	"context"
+	"io"
+	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/benthosdev/benthos/v4/internal/component/scanner/testutil"
@@ -34,6 +38,45 @@ test:
 	)
 }
 
+func TestJSONScannerBadData(t *testing.T) {
+	confSpec := service.NewConfigSpec().Field(service.NewScannerField("test"))
+	pConf, err := confSpec.ParseYAML(`
+test:
+  json_documents: {}
+`, nil)
+	require.NoError(t, err)
+
+	rdr, err := pConf.FieldScanner("test")
+	require.NoError(t, err)
+
+	var ack error
+
+	scanner, err := rdr.Create(io.NopCloser(strings.NewReader(`{"a":"a0"}
+nope !@ not good json
+{"a":"a1"}
+`)), func(ctx context.Context, err error) error {
+		ack = err
+		return nil
+	}, &service.ScannerSourceDetails{})
+	require.NoError(t, err)
+
+	resBatch, aFn, err := scanner.NextBatch(context.Background())
+	require.NoError(t, err)
+	require.NoError(t, aFn(context.Background(), nil))
+	require.Len(t, resBatch, 1)
+	mBytes, err := resBatch[0].AsBytes()
+	require.NoError(t, err)
+	assert.Equal(t, `{"a":"a0"}`, string(mBytes))
+
+	_, _, err = scanner.NextBatch(context.Background())
+	assert.Error(t, err)
+
+	_, _, err = scanner.NextBatch(context.Background())
+	assert.ErrorIs(t, err, io.EOF)
+
+	assert.ErrorContains(t, ack, "invalid character")
+}
+
 func TestJSONScannerFormatted(t *testing.T) {
 	confSpec := service.NewConfigSpec().Field(service.NewScannerField("test"))
 	pConf, err := confSpec.ParseYAML(`

website/docs/components/scanners/json_documents.md

@@ -16,20 +16,11 @@ import TabItem from '@theme/TabItem';
 
 Consumes a stream of one or more JSON documents.
 
+Introduced in version 4.27.0.
+
 ```yml
 # Config fields, showing default values
-json_documents:
-  continue_on_error: false
+json_documents: {}
 ```
 
-## Fields
-
-### `continue_on_error`
-
-If a JSON decoding fails due to any error emit an empty message marked with the error and then continue consuming subsequent documents when possible.
-
-
-Type: `bool`  
-Default: `false`  
-