Add support for interpolated URLs.

Is e.g. necessary for Redis processors/outputs later on.

public/service/config_interpolated_url.go

@@ -0,0 +1,39 @@
+package service
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/redpanda-data/benthos/v4/internal/docs"
+)
+
+// NewInterpolatedURLField defines a new config field that describes a
+// dynamic URL that supports Bloblang interpolation functions. It is then
+// possible to extract an *FieldInterpolatedURL from the resulting parsed config
+// with the method FieldInterpolatedURL.
+func NewInterpolatedURLField(name string) *ConfigField {
+	tf := docs.FieldURL(name, "").IsInterpolated()
+	return &ConfigField{field: tf}
+}
+
+// FieldInterpolatedURL accesses a field from a parsed config that was
+// defined with NewInterpolatedURLField and returns either an
+// *InterpolatedURL or an error if the string was invalid.
+func (p *ParsedConfig) FieldInterpolatedURL(path ...string) (*InterpolatedURL, error) {
+	v, exists := p.i.Field(path...)
+	if !exists {
+		return nil, fmt.Errorf("field '%v' was not found in the config", strings.Join(path, "."))
+	}
+
+	str, ok := v.(string)
+	if !ok {
+		return nil, fmt.Errorf("expected field '%v' to be a string, got %T", strings.Join(path, "."), v)
+	}
+
+	e, err := p.mgr.BloblEnvironment().NewField(str)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse interpolated field '%v': %v", strings.Join(path, "."), err)
+	}
+
+	return &InterpolatedURL{expr: e}, nil
+}

public/service/interpolated_url.go

@@ -0,0 +1,57 @@
+package service
+
+import (
+	"net/url"
+
+	"github.com/redpanda-data/benthos/v4/internal/bloblang"
+	"github.com/redpanda-data/benthos/v4/internal/bloblang/field"
+)
+
+// InterpolatedURL resolves a URL containing dynamic interpolation
+// functions for a given message.
+type InterpolatedURL struct {
+	expr *field.Expression
+}
+
+// NewInterpolatedURL parses an interpolated URL expression.
+func NewInterpolatedURL(expr string) (*InterpolatedURL, error) {
+	e, err := bloblang.GlobalEnvironment().NewField(expr)
+	if err != nil {
+		return nil, err
+	}
+	return &InterpolatedURL{expr: e}, nil
+}
+
+// Static returns the underlying contents of the interpolated URL only if it
+// contains zero dynamic expressions, and is therefore static, otherwise an
+// empty string is returned. A second boolean parameter is also returned
+// indicating whether the URL was static, helping to distinguish between a
+// static empty URL versus a non-static URL.
+func (i *InterpolatedURL) Static() (*url.URL, bool) {
+	if i.expr.NumDynamicExpressions() > 0 {
+		return nil, false
+	}
+	s, _ := i.expr.String(0, nil)
+
+	u, err := url.Parse(s)
+	if err != nil {
+		return nil, false
+	}
+
+	return u, true
+}
+
+// TryURL resolves the interpolated field for a given message as a URL,
+// returns an error if any interpolation functions fail.
+func (i *InterpolatedURL) TryURL(m *Message) (*url.URL, error) {
+	s, err := i.expr.String(0, fauxOldMessage{m.part})
+	if err != nil {
+		return nil, err
+	}
+
+	u, err := url.Parse(s)
+	if err != nil {
+		return nil, err
+	}
+	return u, nil
+}

public/service/interpolated_url_test.go

@@ -0,0 +1,94 @@
+package service
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInterpolatedURL(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		expr     string
+		msg      *Message
+		expected *url.URL
+	}{
+		{
+			name:     "content interpolation",
+			expr:     `http://foo.com/${! content() }/bar`,
+			msg:      NewMessage([]byte("hello world")),
+			expected: mustParseUrl(`http://foo.com/hello world/bar`),
+		},
+		{
+			name:     "no interpolation",
+			expr:     `https://foo.bar`,
+			msg:      NewMessage([]byte("hello world")),
+			expected: mustParseUrl(`https://foo.bar`),
+		},
+		{
+			name: "metadata interpolation",
+			expr: `http://foo.com/${! meta("var1") }/bar`,
+			msg: func() *Message {
+				m := NewMessage([]byte("hello world"))
+				m.MetaSet("var1", "value1")
+				return m
+			}(),
+			expected: mustParseUrl("http://foo.com/value1/bar"),
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+
+		t.Run("api/"+test.name, func(t *testing.T) {
+			t.Parallel()
+
+			i, err := NewInterpolatedURL(test.expr)
+			require.NoError(t, err)
+
+			{
+				got, err := i.TryURL(test.msg)
+				require.NoError(t, err)
+
+				assert.Equal(t, test.expected, got)
+			}
+		})
+	}
+}
+
+func TestInterpolatedURLCtor(t *testing.T) {
+	t.Parallel()
+
+	i, err := NewInterpolatedURL(`http://foo.com/${! meta("var1")  bar`)
+
+	assert.EqualError(t, err, "required: expected end of expression, got: bar")
+	assert.Nil(t, i)
+}
+
+func TestInterpolatedURLMethods(t *testing.T) {
+	t.Parallel()
+
+	i, err := NewInterpolatedURL(`http://foo.com/${! meta("var1") + 1 }/bar`)
+	require.NoError(t, err)
+
+	m := NewMessage([]byte("hello world"))
+	m.MetaSet("var1", "value1")
+
+	{
+		got, err := i.TryURL(m)
+		require.EqualError(t, err, "cannot add types string (from meta field var1) and number (from number literal)")
+		require.Empty(t, got)
+	}
+}
+
+func mustParseUrl(s string) *url.URL {
+	u, err := url.Parse(s)
+	if err != nil {
+		panic(err)
+	}
+	return u
+}