Add asyncConnect and TimeOut to netty Transceiver

Signed-off-by: Sébastien GLON <[email protected]>

examples/flume/client.go

@@ -1,21 +1,17 @@
 package main
 import (
 	"github.com/sebglon/goavro"
-	"net"
 	"log"
-
+	"github.com/sebglon/goavro/transceiver/netty"
+	"time"
 )
 
 func main() {
 	//t.SkipNow()
-	rAddr, err := net.ResolveTCPAddr("tcp", "10.98.80.113:63001")
-	conn, err := net.DialTCP("tcp", nil, rAddr)
+	transceiver,err := netty.NewTransceiver(netty.Config{AsyncConnect:true, NettyHost:"10.98.80.113"})
 	if err != nil {
 		log.Fatal(err)
 	}
-	defer conn.Close()
-
-	transceiver := goavro.NewNettyTransceiver(conn)
 	protocol, err := goavro.NewProtocol()
 	if err != nil {
 		log.Fatal(err)
@@ -30,12 +26,14 @@ func main() {
 	flumeRecord.Set("headers", headers)
 	flumeRecord.Set("body", []byte("2016-08-02 14:45:38|flume.composantTechnique_IS_UNDEFINED|flume.application_IS_UNDEFINED|flume.client_IS_UNDEFINED|flume.plateforme_IS_UNDEFINED|instance_IS_UNDEFINED|logname_IS_UNDEFINED|WARN |test.LogGenerator|test !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"))
 	requestor := goavro.NewRequestor(protocol, transceiver)
+
+
 	err = requestor.Request("append", flumeRecord)
 
 	if err != nil {
 		log.Fatal("Request: ", err)
 	}
-
+	time.Sleep(5 * time.Second)
 	err = requestor.Request("append", flumeRecord)
 
 	if err != nil {

protocol_test.go

@@ -98,7 +98,6 @@ func TestGetCodec(t *testing.T) {
 		t.Fatal(errFlume)
 	}
 	headers := make(map[string]interface{})
-	headers[AVRO_SCHEMA_LITERAL_HEADER] = stringSchema
 	headers["host_header"] = "127.0.0.1"
 	flumeRecord.Set("headers", headers)
 	flumeRecord.Set("body", []byte("test"))

requestor.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"github.com/sebglon/goavro/transceiver"
 )
 
 var REMOTE_HASHES map[string][]byte
@@ -20,7 +21,7 @@ var HANDSHAKE_REQUESTOR_READER Codec
 type Requestor struct {
 	// Base class for the client side of protocol interaction.
 	local_protocol		Protocol
-	transceiver		Transceiver
+	transceiver		transceiver.Transceiver
 	remote_protocol 	Protocol
 	remote_hash		[]byte
 	send_protocol		bool
@@ -43,7 +44,7 @@ func init() {
 
 }
 
-func NewRequestor(localProto Protocol, transceiver Transceiver) *Requestor {
+func NewRequestor(localProto Protocol, transceiver transceiver.Transceiver) *Requestor {
 	return &Requestor{
 		local_protocol: localProto,
 		transceiver: transceiver,
@@ -57,12 +58,12 @@ func NewRequestor(localProto Protocol, transceiver Transceiver) *Requestor {
 
 func (a *Requestor) RemoteProtocol(proto Protocol) {
 	a.remote_protocol = proto
-	REMOTE_PROTOCOLS[a.transceiver.RemoteName()] = proto
+	REMOTE_PROTOCOLS[proto.Name] = proto
 }
 
 func (a *Requestor) RemoteHash(hash []byte) {
 	a.remote_hash =  hash
-	REMOTE_HASHES[a.transceiver.RemoteName()] = hash
+	REMOTE_HASHES[a.remote_protocol.Name] = hash
 }
 
 func (a *Requestor) Request(message_name string, request_datum  interface{})  error {
@@ -94,18 +95,21 @@ func (a *Requestor) Request(message_name string, request_datum  interface{})  er
 	}
 	//buffer_decoder := bytes.NewBuffer(decoder)
 	// process the handshake and call response
-	ok, err := a.read_handshake_response(responses[0])
-	if err!=nil {
-		return err
-	}
-	a.send_handshake= !ok
 
-	if ok {
-		a.read_call_responseCode(responses[1])
-		if err!=nil {
+	if len(responses) >0 {
+		ok, err := a.read_handshake_response(responses[0])
+		if err != nil {
 			return err
 		}
-	//	a.Request(message_name, request_datum)
+		a.send_handshake = !ok
+
+		if ok {
+			a.read_call_responseCode(responses[1])
+			if err != nil {
+				return err
+			}
+			//	a.Request(message_name, request_datum)
+		}
 	}
 	return nil
 }

requestor_test.go

@@ -5,6 +5,7 @@ import (
 	"net"
 	"bytes"
 	"reflect"
+	netty "github.com/sebglon/goavro/transceiver/netty"
 )
 
 func TestWrite_handshake_request(t *testing.T) {
@@ -16,7 +17,10 @@ func TestWrite_handshake_request(t *testing.T) {
 	}
 	defer conn.Close()
 
-	transceiver := NewNettyTransceiver(conn)
+	transceiver, err := netty.NewTransceiver(netty.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
 	protocol, err := NewProtocol()
 	if err != nil {
 		t.Fatal(err)
@@ -75,14 +79,12 @@ func TestRead_handshake_reponse(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	rAddr, err := net.ResolveTCPAddr("tcp", "10.98.80.113:63001")
-	conn, err := net.DialTCP("tcp", nil, rAddr)
+
+	transceiver, err := netty.NewTransceiver(netty.Config{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer conn.Close()
 
-	transceiver := NewNettyTransceiver(conn)
 	protocol, err := NewProtocol()
 	if err != nil {
 		t.Fatal(err)
@@ -96,17 +98,21 @@ func TestRead_handshake_reponse(t *testing.T) {
 
 }
 
+type Conn struct {
+	bytes.Buffer
+}
+func (c *Conn) Close() error {
+	return nil
+}
 
 func TestWrite_call_request(t *testing.T) {
 	//t.SkipNow()
-	rAddr, err := net.ResolveTCPAddr("tcp", "10.98.80.113:63001")
-	conn, err := net.DialTCP("tcp", nil, rAddr)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer conn.Close()
 
-	transceiver := NewNettyTransceiver(conn)
+
+	transceiver, err := netty.NewTransceiver(netty.Config{})
+	buf := &Conn{}
+	transceiver.Conn = buf
+
 	protocol, err := NewProtocol()
 	if err != nil {
 		t.Fatal(err)
@@ -147,14 +153,10 @@ func TestWrite_call_request(t *testing.T) {
 
 func TestWrite_call_requestHeader(t *testing.T) {
 	//t.SkipNow()
-	rAddr, err := net.ResolveTCPAddr("tcp", "10.98.80.113:63001")
-	conn, err := net.DialTCP("tcp", nil, rAddr)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer conn.Close()
+	transceiver, err := netty.NewTransceiver(netty.Config{})
+	buf := &Conn{}
+	transceiver.Conn = buf
 
-	transceiver := NewNettyTransceiver(conn)
 	protocol, err := NewProtocol()
 	if err != nil {
 		t.Fatal(err)
@@ -173,16 +175,12 @@ func TestWrite_call_requestHeader(t *testing.T) {
 }
 
 func TestRead_call_responseMessage(t *testing.T) {
-	//t.SkipNow()
 
-	rAddr, err := net.ResolveTCPAddr("tcp", "10.98.80.113:63001")
-	conn, err := net.DialTCP("tcp", nil, rAddr)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer conn.Close()
 
-	transceiver := NewNettyTransceiver(conn)
+	transceiver, err := netty.NewTransceiver(netty.Config{})
+	buf := &Conn{}
+	transceiver.Conn = buf
+
 	protocol, err := NewProtocol()
 	if err != nil {
 		t.Fatal(err)

transceiver/Tranceiver.go

@@ -0,0 +1,11 @@
+package transceiver
+
+import (
+	"bytes"
+	"io"
+)
+
+type Transceiver interface {
+	Transceive(request []bytes.Buffer) ([]io.Reader, error)
+
+}