From c1ff3ceee423bc7a747e20bccf8271108aea45e8 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Fri, 9 Aug 2019 12:31:29 +0100 Subject: [PATCH 1/6] Add more verbose not found error --- client/grpc/grpc.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/client/grpc/grpc.go b/client/grpc/grpc.go index c6a3f5dd..7ae85347 100644 --- a/client/grpc/grpc.go +++ b/client/grpc/grpc.go @@ -74,7 +74,7 @@ func (g *grpcClient) next(request client.Request, opts client.CallOptions) (sele // get next nodes from the selector next, err := g.opts.Selector.Select(service, opts.SelectOptions...) if err != nil && err == selector.ErrNotFound { - return nil, errors.NotFound("go.micro.client", err.Error()) + return nil, errors.NotFound("go.micro.client", "service %s not found: %v", service, err.Error()) } else if err != nil { return nil, errors.InternalServerError("go.micro.client", err.Error()) } @@ -351,7 +351,7 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface // select next node node, err := next() if err != nil && err == selector.ErrNotFound { - return errors.NotFound("go.micro.client", err.Error()) + return errors.NotFound("go.micro.client", "service %s not found: %v", req.Service(), err.Error()) } else if err != nil { return errors.InternalServerError("go.micro.client", err.Error()) } @@ -430,7 +430,7 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli node, err := next() if err != nil && err == selector.ErrNotFound { - return nil, errors.NotFound("go.micro.client", err.Error()) + return nil, errors.NotFound("go.micro.client", "service %s not found: %v", req.Service(), err.Error()) } else if err != nil { return nil, errors.InternalServerError("go.micro.client", err.Error()) } From 2c66e94045c02415ff08295093e409f4518e4d23 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Sat, 10 Aug 2019 16:37:49 +0100 Subject: [PATCH 2/6] fix some tunnel bugs like races and duplicate messages... --- tunnel/default.go | 20 +++++++++++++------- tunnel/listener.go | 3 --- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/tunnel/default.go b/tunnel/default.go index 842d0a95..e5f09708 100644 --- a/tunnel/default.go +++ b/tunnel/default.go @@ -111,25 +111,26 @@ func (t *tun) process() { for { select { case msg := <-t.send: - nmsg := &transport.Message{ - Header: msg.data.Header, + newMsg := &transport.Message{ + Header: make(map[string]string), Body: msg.data.Body, } - if nmsg.Header == nil { - nmsg.Header = make(map[string]string) + for k, v := range msg.data.Header { + newMsg.Header[k] = v } // set the tunnel id on the outgoing message - nmsg.Header["Micro-Tunnel-Id"] = msg.id + newMsg.Header["Micro-Tunnel-Id"] = msg.id // set the session id - nmsg.Header["Micro-Tunnel-Session"] = msg.session + newMsg.Header["Micro-Tunnel-Session"] = msg.session // send the message via the interface t.RLock() for _, link := range t.links { - link.Send(nmsg) + log.Debugf("Sending %+v to %s", newMsg, link.Remote()) + link.Send(newMsg) } t.RUnlock() case <-t.closed: @@ -170,6 +171,7 @@ func (t *tun) listen(link transport.Socket, listener bool) { var s *socket var exists bool + log.Debugf("Received %+v from %s", msg, link.Remote()) // if its a local listener then we use that as the session id // e.g we're using a loopback connecting to ourselves if listener { @@ -189,10 +191,12 @@ func (t *tun) listen(link transport.Socket, listener bool) { // no socket in existence if !exists { + log.Debugf("Skipping") // drop it, we don't care about // messages we don't know about continue } + log.Debugf("Using socket %s %s", s.id, s.session) // is the socket closed? select { @@ -398,6 +402,7 @@ func (t *tun) Init(opts ...Option) error { // Dial an address func (t *tun) Dial(addr string) (Conn, error) { + log.Debugf("Tunnel dialing %s", addr) c, ok := t.newSocket(addr, t.newSession()) if !ok { return nil, errors.New("error dialing " + addr) @@ -413,6 +418,7 @@ func (t *tun) Dial(addr string) (Conn, error) { // Accept a connection on the address func (t *tun) Listen(addr string) (Listener, error) { + log.Debugf("Tunnel listening on %s", addr) // create a new socket by hashing the address c, ok := t.newSocket(addr, "listener") if !ok { diff --git a/tunnel/listener.go b/tunnel/listener.go index 070b313b..368cf4a5 100644 --- a/tunnel/listener.go +++ b/tunnel/listener.go @@ -48,9 +48,6 @@ func (t *tunListener) process() { wait: make(chan bool), } - // first message - sock.recv <- m - // save the socket conns[m.session] = sock From 6dd3ea1853c77a53253988a1fa48abc9291647b2 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Sat, 10 Aug 2019 18:44:50 +0100 Subject: [PATCH 3/6] Remove listen check --- tunnel/default.go | 23 +++++++++-------------- 1 file changed, 9 insertions(+), 14 deletions(-) diff --git a/tunnel/default.go b/tunnel/default.go index e5f09708..6f7fd2c1 100644 --- a/tunnel/default.go +++ b/tunnel/default.go @@ -128,6 +128,7 @@ func (t *tun) process() { // send the message via the interface t.RLock() + log.Debugf("Sending some shit %d links", len(t.links)) for _, link := range t.links { log.Debugf("Sending %+v to %s", newMsg, link.Remote()) link.Send(newMsg) @@ -172,21 +173,15 @@ func (t *tun) listen(link transport.Socket, listener bool) { var exists bool log.Debugf("Received %+v from %s", msg, link.Remote()) - // if its a local listener then we use that as the session id - // e.g we're using a loopback connecting to ourselves - if listener { + // get the socket based on the tunnel id and session + // this could be something we dialed in which case + // we have a session for it otherwise its a listener + s, exists = t.getSocket(id, session) + if !exists { + // try get it based on just the tunnel id + // the assumption here is that a listener + // has no session but its set a listener session s, exists = t.getSocket(id, "listener") - } else { - // get the socket based on the tunnel id and session - // this could be something we dialed in which case - // we have a session for it otherwise its a listener - s, exists = t.getSocket(id, session) - if !exists { - // try get it based on just the tunnel id - // the assumption here is that a listener - // has no session but its set a listener session - s, exists = t.getSocket(id, "listener") - } } // no socket in existence From 8986b3135f72770b70596871e4ecd8c508c7b354 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Sat, 10 Aug 2019 18:46:54 +0100 Subject: [PATCH 4/6] Strip logging --- tunnel/default.go | 1 - 1 file changed, 1 deletion(-) diff --git a/tunnel/default.go b/tunnel/default.go index 6f7fd2c1..480fd465 100644 --- a/tunnel/default.go +++ b/tunnel/default.go @@ -128,7 +128,6 @@ func (t *tun) process() { // send the message via the interface t.RLock() - log.Debugf("Sending some shit %d links", len(t.links)) for _, link := range t.links { log.Debugf("Sending %+v to %s", newMsg, link.Remote()) link.Send(newMsg) From e613b0c2058d31d8e22daa3162a1bbcaaf6d6923 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Sun, 11 Aug 2019 09:54:02 +0100 Subject: [PATCH 5/6] remove the single node tunnel test --- tunnel/tunnel_test.go | 22 ---------------------- 1 file changed, 22 deletions(-) diff --git a/tunnel/tunnel_test.go b/tunnel/tunnel_test.go index 1580d4a9..721479bb 100644 --- a/tunnel/tunnel_test.go +++ b/tunnel/tunnel_test.go @@ -54,28 +54,6 @@ func testSend(t *testing.T, tun Tunnel) { } func TestTunnel(t *testing.T) { - // create a new listener - tun := NewTunnel(Nodes("127.0.0.1:9096")) - err := tun.Connect() - if err != nil { - t.Fatal(err) - } - defer tun.Close() - - var wg sync.WaitGroup - - // start accepting connections - wg.Add(1) - go testAccept(t, tun, &wg) - - // send a message - testSend(t, tun) - - // wait until message is received - wg.Wait() -} - -func TestTwoTunnel(t *testing.T) { // create a new tunnel client tunA := NewTunnel( Address("127.0.0.1:9096"), From cd2ac648ffc008e755f0c5d6a509e82d8a451686 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=88=98=E5=B0=8F=E4=B9=90?= Date: Sun, 11 Aug 2019 18:05:35 +0800 Subject: [PATCH 6/6] Fix read yaml config from memory package main import ( "fmt" "github.com/micro/go-micro/config" "github.com/micro/go-micro/config/source/memory" ) var configData = []byte(` --- a: 1234 `) func main() { memorySource := memory.NewSource( memory.WithYAML(configData), ) // Create new config conf := config.NewConfig() // Load file source conf.Load(memorySource) fmt.Println(string(conf.Bytes())) } --- config/source/memory/memory.go | 1 + 1 file changed, 1 insertion(+) diff --git a/config/source/memory/memory.go b/config/source/memory/memory.go index 607c7b4a..78e26d3f 100644 --- a/config/source/memory/memory.go +++ b/config/source/memory/memory.go @@ -18,6 +18,7 @@ type memory struct { func (s *memory) Read() (*source.ChangeSet, error) { s.RLock() cs := &source.ChangeSet{ + Format: s.ChangeSet.Format, Timestamp: s.ChangeSet.Timestamp, Data: s.ChangeSet.Data, Checksum: s.ChangeSet.Checksum,