Introduce HTTPPayload struct and HTTPPayloader interface to MarshalJSON() all the data structures that are returned by the HTTP protocol

tap/extensions/http/matcher.go

@@ -2,12 +2,13 @@ package main
 
 import (
 	"fmt"
-	"github.com/romana/rlog"
 	"net/http"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/romana/rlog"
+
 	"github.com/up9inc/mizu/tap/api"
 )
 
@@ -31,7 +32,10 @@ func (matcher *requestResponseMatcher) registerRequest(ident string, request *ht
 	requestHTTPMessage := api.GenericMessage{
 		IsRequest:   true,
 		CaptureTime: captureTime,
-		Orig:        request,
+		Orig: HTTPPayload{
+			Type: "http_request",
+			Data: request,
+		},
 	}
 
 	if response, found := matcher.openMessagesMap.LoadAndDelete(key); found {
@@ -58,7 +62,10 @@ func (matcher *requestResponseMatcher) registerResponse(ident string, response *
 	responseHTTPMessage := api.GenericMessage{
 		IsRequest:   false,
 		CaptureTime: captureTime,
-		Orig:        response,
+		Orig: HTTPPayload{
+			Type: "http_response",
+			Data: response,
+		},
 	}
 
 	if request, found := matcher.openMessagesMap.LoadAndDelete(key); found {

tap/extensions/http/structs.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"fmt"
+)
+
+type HTTPPayload struct {
+	Type string
+	Data interface{}
+}
+
+type HTTPPayloader interface {
+	MarshalJSON() ([]byte, error)
+}
+
+func (h HTTPPayload) MarshalJSON() ([]byte, error) {
+	switch h.Type {
+	case "http_request":
+		return []byte("{\"val\": \"" + h.Type + "\"}"), nil
+	case "http_response":
+		return []byte("{\"val\": \"" + h.Type + "\"}"), nil
+	default:
+		panic(fmt.Sprintf("HTTP payload cannot be marshaled: %s\n", h.Type))
+	}
+}

tap/tcp_reader.go

@@ -4,7 +4,6 @@ import (
 	"bufio"
 	"fmt"
 	"io"
-	"log"
 	"strconv"
 	"sync"
 	"time"
@@ -103,7 +102,7 @@ func containsPort(ports []string, port string) bool {
 
 func (h *tcpReader) run(wg *sync.WaitGroup) {
 	defer wg.Done()
-	log.Printf("Called run h.isClient: %v\n", h.isClient)
+	// log.Printf("Called run h.isClient: %v\n", h.isClient)
 	b := bufio.NewReader(h)
 	if h.isClient {
 		extensions[1].Dissector.Dissect(b, h.isClient, h.tcpID, h.Emitter)

tap/tcp_stream_factory.go

@@ -2,7 +2,6 @@ package tap
 
 import (
 	"fmt"
-	"log"
 	"sync"
 
 	"github.com/romana/rlog"
@@ -26,7 +25,7 @@ type tcpStreamFactory struct {
 }
 
 func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream {
-	log.Printf("* NEW: %s %s", net, transport)
+	rlog.Debugf("* NEW: %s %s", net, transport)
 	fsmOptions := reassembly.TCPSimpleFSMOptions{
 		SupportMissingEstablishment: *allowmissinginit,
 	}