Implemented streaming writer (#621)

* Implemented streaming writer

Signed-off-by: Francesco Guardiani <[email protected]>

* Little doc fix

Signed-off-by: Francesco Guardiani <[email protected]>

* Move is base 64 check

Signed-off-by: Francesco Guardiani <[email protected]>

* Test

Signed-off-by: Francesco Guardiani <[email protected]>

* Stream from the pool

Signed-off-by: Francesco Guardiani <[email protected]>

Author: slinkydeveloper

v2/binding/format/format_test.go

@@ -1,6 +1,7 @@
 package format_test
 
 import (
+	"encoding/json"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -23,7 +24,15 @@ func TestJSON(t *testing.T) {
 	require.NoError(e.SetData(event.ApplicationJSON, "foo"))
 	b, err := format.JSON.Marshal(&e)
 	require.NoError(err)
-	require.Equal(`{"data":"foo","datacontenttype":"application/json","ex":"val","id":"id","source":"source","specversion":"0.3","type":"type"}`, string(b))
+	assertJsonEquals(t, map[string]interface{}{
+		"data":            "foo",
+		"datacontenttype": "application/json",
+		"ex":              "val",
+		"id":              "id",
+		"source":          "source",
+		"specversion":     "0.3",
+		"type":            "type",
+	}, b)
 
 	var e2 event.Event
 	require.NoError(format.JSON.Unmarshal(b, &e2))
@@ -65,7 +74,14 @@ func TestMarshalUnmarshal(t *testing.T) {
 	require.NoError(e.SetData(event.ApplicationJSON, "foo"))
 	b, err := format.Marshal(format.JSON.MediaType(), &e)
 	require.NoError(err)
-	require.Equal(`{"data":"foo","datacontenttype":"application/json","id":"id","source":"source","specversion":"0.3","type":"type"}`, string(b))
+	assertJsonEquals(t, map[string]interface{}{
+		"data":            "foo",
+		"datacontenttype": "application/json",
+		"id":              "id",
+		"source":          "source",
+		"specversion":     "0.3",
+		"type":            "type",
+	}, b)
 
 	var e2 event.Event
 	require.NoError(format.Unmarshal(format.JSON.MediaType(), b, &e2))
@@ -99,3 +115,16 @@ func TestAdd(t *testing.T) {
 	require.NoError(err)
 	require.Equal([]byte("undummy!"), e.Data())
 }
+
+func assertJsonEquals(t *testing.T, want map[string]interface{}, got []byte) {
+	var gotToCompare map[string]interface{}
+	require.NoError(t, json.Unmarshal(got, &gotToCompare))
+
+	// Marshal and unmarshal want to make sure the types are correct
+	wantBytes, err := json.Marshal(want)
+	require.NoError(t, err)
+	var wantToCompare map[string]interface{}
+	require.NoError(t, json.Unmarshal(wantBytes, &wantToCompare))
+
+	require.Equal(t, wantToCompare, gotToCompare)
+}

v2/client/client_test.go

@@ -3,6 +3,7 @@ package client_test
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"log"
@@ -81,6 +82,19 @@ func simpleStructuredClient(target string) client.Client {
 func TestClientSend(t *testing.T) {
 	now := time.Now()
 
+	structuredEvent := event.Event{
+		Context: event.EventContextV03{
+			Type:   "unit.test.client",
+			Source: *types.ParseURIRef("/unit/test/client"),
+			Time:   &types.Timestamp{Time: now},
+			ID:     "AABBCCDDEE",
+		}.AsV03(),
+	}
+	_ = structuredEvent.SetData(event.ApplicationJSON, &map[string]interface{}{
+		"sq":  42,
+		"msg": "hello",
+	})
+
 	testCases := map[string]struct {
 		c       func(target string) client.Client
 		event   event.Event
@@ -121,32 +135,19 @@ func TestClientSend(t *testing.T) {
 			},
 		},
 		"structured simple v0.3": {
-			c: simpleStructuredClient,
-			event: func() event.Event {
-				e := event.Event{
-					Context: event.EventContextV03{
-						Type:   "unit.test.client",
-						Source: *types.ParseURIRef("/unit/test/client"),
-						Time:   &types.Timestamp{Time: now},
-						ID:     "AABBCCDDEE",
-					}.AsV03(),
-				}
-				_ = e.SetData(event.ApplicationJSON, &map[string]interface{}{
-					"sq":  42,
-					"msg": "hello",
-				})
-				return e
-			}(),
+			c:     simpleStructuredClient,
+			event: structuredEvent,
 			resp: &http.Response{
 				StatusCode: http.StatusAccepted,
 			},
 			want: &requestValidation{
 				Headers: map[string][]string{
 					"content-type": {"application/cloudevents+json"},
 				},
-				Body: []byte(fmt.Sprintf(`{"data":{"msg":"hello","sq":42},"datacontenttype":"application/json","id":"AABBCCDDEE","source":"/unit/test/client","specversion":"0.3","time":%q,"type":"unit.test.client"}`,
-					now.UTC().Format(time.RFC3339Nano)),
-				),
+				Body: func() []byte {
+					b, _ := json.Marshal(structuredEvent)
+					return b
+				}(),
 			},
 		},
 	}