diff --git a/tap/api/api.go b/tap/api/api.go index 77151ca33..456926948 100644 --- a/tap/api/api.go +++ b/tap/api/api.go @@ -1,6 +1,9 @@ package api -import "plugin" +import ( + "bufio" + "plugin" +) type Extension struct { Name string @@ -13,4 +16,5 @@ type Extension struct { type Dissector interface { Register(*Extension) Ping() + Dissect(b *bufio.Reader) interface{} } diff --git a/tap/extensions/amqp/main.go b/tap/extensions/amqp/main.go index 37372164f..72ff97624 100644 --- a/tap/extensions/amqp/main.go +++ b/tap/extensions/amqp/main.go @@ -1,7 +1,7 @@ package main import ( - "fmt" + "bufio" "log" "github.com/up9inc/mizu/tap/api" @@ -19,7 +19,12 @@ func (g dissecting) Register(extension *api.Extension) { } func (g dissecting) Ping() { - fmt.Printf("pong AMQP\n") + log.Printf("pong AMQP\n") +} + +func (g dissecting) Dissect(b *bufio.Reader) interface{} { + // TODO: Implement + return nil } // exported as symbol named "Greeter" diff --git a/tap/extensions/http/main.go b/tap/extensions/http/main.go index ef108df4e..198fe51e0 100644 --- a/tap/extensions/http/main.go +++ b/tap/extensions/http/main.go @@ -1,8 +1,9 @@ package main import ( - "fmt" + "bufio" "log" + "net/http" "github.com/up9inc/mizu/tap/api" ) @@ -19,7 +20,14 @@ func (g dissecting) Register(extension *api.Extension) { } func (g dissecting) Ping() { - fmt.Printf("pong HTTP\n") + log.Printf("pong HTTP\n") +} + +func (g dissecting) Dissect(b *bufio.Reader) interface{} { + log.Printf("called Dissect!") + req, _ := http.ReadRequest(b) + log.Printf("HTTP Request: %+v\n", req) + return nil } // exported as symbol named "Greeter" diff --git a/tap/extensions/kafka/main.go b/tap/extensions/kafka/main.go index db06c651f..d9ae9d174 100644 --- a/tap/extensions/kafka/main.go +++ b/tap/extensions/kafka/main.go @@ -1,7 +1,7 @@ package main import ( - "fmt" + "bufio" "log" "github.com/up9inc/mizu/tap/api" @@ -19,7 +19,12 @@ func (g dissecting) Register(extension *api.Extension) { } func (g dissecting) Ping() { - fmt.Printf("pong Kafka\n") + log.Printf("pong Kafka\n") +} + +func (g dissecting) Dissect(b *bufio.Reader) interface{} { + // TODO: Implement + return nil } // exported as symbol named "Greeter" diff --git a/tap/tcp_stream_factory.go b/tap/tcp_stream_factory.go index 76dd8e22b..370f6434a 100644 --- a/tap/tcp_stream_factory.go +++ b/tap/tcp_stream_factory.go @@ -1,6 +1,7 @@ package tap import ( + "bufio" "fmt" "io" "sync" @@ -39,7 +40,9 @@ func (h *tcpStream) run(wg *sync.WaitGroup) { defer wg.Done() for _, extension := range extensions { if containsPort(extension.Ports, h.transport.Dst().String()) { + b := bufio.NewReader(h) extension.Dissector.Ping() + extension.Dissector.Dissect(b) } } // b := bufio.NewReader(h)