1 files changed, 120 insertions, 115 deletions

diff --git a/src/gnunet/service/gns/service.go b/src/gnunet/service/gns/service.go
index 6fa1eb2..326d831 100644
--- a/src/gnunet/service/gns/service.go
+++ b/src/gnunet/service/gns/service.go
@@ -19,6 +19,7 @@
 package gns
 
 import (
+        "context"
         "encoding/hex"
         "fmt"
         "io"
@@ -28,8 +29,8 @@ import (
         "gnunet/enums"
         "gnunet/message"
         "gnunet/service"
+        "gnunet/service/dht/blocks"
         "gnunet/service/revocation"
-        "gnunet/transport"
         "gnunet/util"
 
         "github.com/bfix/gospel/data"
@@ -64,115 +65,117 @@ func NewService() service.Service {
         return inst
 }
 
-// Start the GNS service
-func (s *Service) Start(spec string) error {
-        return nil
-}
-
-// Stop the GNS service
-func (s *Service) Stop() error {
-        return nil
-}
-
 // ServeClient processes a client channel.
-func (s *Service) ServeClient(ctx *service.SessionContext, mc *transport.MsgChannel) {
+func (s *Service) ServeClient(ctx context.Context, id int, mc *service.Connection) {
         reqID := 0
-loop:
+        var cancel context.CancelFunc
+        ctx, cancel = context.WithCancel(ctx)
+
         for {
                 // receive next message from client
                 reqID++
-                logger.Printf(logger.DBG, "[gns:%d:%d] Waiting for client request...\n", ctx.ID, reqID)
-                msg, err := mc.Receive(ctx.Signaller())
+                logger.Printf(logger.DBG, "[gns:%d:%d] Waiting for client request...\n", id, reqID)
+                msg, err := mc.Receive(ctx)
                 if err != nil {
                         if err == io.EOF {
-                                logger.Printf(logger.INFO, "[gns:%d:%d] Client channel closed.\n", ctx.ID, reqID)
-                        } else if err == transport.ErrChannelInterrupted {
-                                logger.Printf(logger.INFO, "[gns:%d:%d] Service operation interrupted.\n", ctx.ID, reqID)
+                                logger.Printf(logger.INFO, "[gns:%d:%d] Client channel closed.\n", id, reqID)
+                        } else if err == service.ErrConnectionInterrupted {
+                                logger.Printf(logger.INFO, "[gns:%d:%d] Service operation interrupted.\n", id, reqID)
                         } else {
-                                logger.Printf(logger.ERROR, "[gns:%d:%d] Message-receive failed: %s\n", ctx.ID, reqID, err.Error())
+                                logger.Printf(logger.ERROR, "[gns:%d:%d] Message-receive failed: %s\n", id, reqID, err.Error())
                         }
-                        break loop
+                        break
                 }
-                logger.Printf(logger.INFO, "[gns:%d:%d] Received request: %v\n", ctx.ID, reqID, msg)
-
-                // perform lookup
-                switch m := msg.(type) {
-                case *message.LookupMsg:
-                        //----------------------------------------------------------
-                        // GNS_LOOKUP
-                        //----------------------------------------------------------
-
-                        // perform lookup on block (locally and remote)
-                        go func(id int, m *message.LookupMsg) {
-                                logger.Printf(logger.INFO, "[gns:%d:%d] Lookup request received.\n", ctx.ID, id)
-                                resp := message.NewGNSLookupResultMsg(m.ID)
-                                ctx.Add()
-                                defer func() {
-                                        // send response
-                                        if resp != nil {
-                                                if err := mc.Send(resp, ctx.Signaller()); err != nil {
-                                                        logger.Printf(logger.ERROR, "[gns:%d:%d] Failed to send response: %s\n", ctx.ID, id, err.Error())
-                                                }
-                                        }
-                                        // go-routine finished
-                                        logger.Printf(logger.DBG, "[gns:%d:%d] Lookup request finished.\n", ctx.ID, id)
-                                        ctx.Remove()
-                                }()
-
-                                label := m.GetName()
-                                kind := NewRRTypeList(int(m.Type))
-                                recset, err := s.Resolve(ctx, label, m.Zone, kind, int(m.Options), 0)
-                                if err != nil {
-                                        logger.Printf(logger.ERROR, "[gns:%d:%d] Failed to lookup block: %s\n", ctx.ID, id, err.Error())
-                                        if err == transport.ErrChannelInterrupted {
-                                                resp = nil
+                logger.Printf(logger.INFO, "[gns:%d:%d] Received request: %v\n", id, reqID, msg)
+
+                // handle message
+                s.HandleMessage(context.WithValue(ctx, "label", fmt.Sprintf(":%d:%d", id, reqID)), msg, mc)
+        }
+        // close client connection
+        mc.Close()
+
+        // cancel all tasks running for this session/connection
+        logger.Printf(logger.INFO, "[gns:%d] Start closing session...\n", id)
+        cancel()
+}
+
+// Handle a single incoming message
+func (s *Service) HandleMessage(ctx context.Context, msg message.Message, back service.Responder) bool {
+        // assemble log label
+        label := ""
+        if v := ctx.Value("label"); v != nil {
+                label = v.(string)
+        }
+        // perform lookup
+        switch m := msg.(type) {
+        case *message.LookupMsg:
+                //----------------------------------------------------------
+                // GNS_LOOKUP
+                //----------------------------------------------------------
+
+                // perform lookup on block (locally and remote)
+                go func(m *message.LookupMsg) {
+                        logger.Printf(logger.INFO, "[gns%s] Lookup request received.\n", label)
+                        resp := message.NewGNSLookupResultMsg(m.ID)
+                        defer func() {
+                                // send response
+                                if resp != nil {
+                                        if err := back.Send(ctx, resp); err != nil {
+                                                logger.Printf(logger.ERROR, "[gns%s] Failed to send response: %s\n", label, err.Error())
                                         }
+                                }
+                                // go-routine finished
+                                logger.Printf(logger.DBG, "[gns%s] Lookup request finished.\n", label)
+                        }()
+
+                        label := m.GetName()
+                        kind := NewRRTypeList(int(m.Type))
+                        recset, err := s.Resolve(ctx, label, m.Zone, kind, int(m.Options), 0)
+                        if err != nil {
+                                logger.Printf(logger.ERROR, "[gns%s] Failed to lookup block: %s\n", label, err.Error())
+                                if err == service.ErrConnectionInterrupted {
+                                        resp = nil
+                                }
+                                return
+                        }
+                        // handle records
+                        if recset != nil {
+                                logger.Printf(logger.DBG, "[gns%s] Received record set with %d entries\n", label, recset.Count)
+
+                                // get records from block
+                                if recset.Count == 0 {
+                                        logger.Printf(logger.WARN, "[gns%s] No records in block\n", label)
                                         return
                                 }
-                                // handle records
-                                if recset != nil {
-                                        logger.Printf(logger.DBG, "[gns:%d:%d] Received record set with %d entries\n", ctx.ID, id, recset.Count)
-
-                                        // get records from block
-                                        if recset.Count == 0 {
-                                                logger.Printf(logger.WARN, "[gns:%d:%d] No records in block\n", ctx.ID, id)
-                                                return
-                                        }
-                                        // process records
-                                        for i, rec := range recset.Records {
-                                                logger.Printf(logger.DBG, "[gns:%d:%d] Record #%d: %v\n", ctx.ID, id, i, rec)
-
-                                                // is this the record type we are looking for?
-                                                if rec.Type == m.Type || int(m.Type) == enums.GNS_TYPE_ANY {
-                                                        // add it to the response message
-                                                        resp.AddRecord(rec)
-                                                }
+                                // process records
+                                for i, rec := range recset.Records {
+                                        logger.Printf(logger.DBG, "[gns%s] Record #%d: %v\n", label, i, rec)
+
+                                        // is this the record type we are looking for?
+                                        if rec.Type == m.Type || int(m.Type) == enums.GNS_TYPE_ANY {
+                                                // add it to the response message
+                                                resp.AddRecord(rec)
                                         }
                                 }
-                        }(reqID, m)
-
-                default:
-                        //----------------------------------------------------------
-                        // UNKNOWN message type received
-                        //----------------------------------------------------------
-                        logger.Printf(logger.ERROR, "[gns:%d:%d] Unhandled message of type (%d)\n", ctx.ID, reqID, msg.Header().MsgType)
-                        break loop
-                }
-        }
-        // close client connection
-        mc.Close()
+                        }
+                }(m)
 
-        // cancel all tasks running for this session/connection
-        logger.Printf(logger.INFO, "[gns:%d] Start closing session... [%d]\n", ctx.ID, ctx.Waiting())
-        ctx.Cancel()
+        default:
+                //----------------------------------------------------------
+                // UNKNOWN message type received
+                //----------------------------------------------------------
+                logger.Printf(logger.ERROR, "[gns%s] Unhandled message of type (%d)\n", label, msg.Header().MsgType)
+                return false
+        }
+        return true
 }
 
 //======================================================================
-// Revocationrelated methods
+// Revocation-related methods
 //======================================================================
 
 // QueryKeyRevocation checks if a key has been revoked
-func (s *Service) QueryKeyRevocation(ctx *service.SessionContext, zkey *crypto.ZoneKey) (valid bool, err error) {
+func (s *Service) QueryKeyRevocation(ctx context.Context, zkey *crypto.ZoneKey) (valid bool, err error) {
         logger.Printf(logger.DBG, "[gns] QueryKeyRev(%s)...\n", util.EncodeBinaryToString(zkey.Bytes()))
 
         // assemble request
@@ -180,7 +183,7 @@ func (s *Service) QueryKeyRevocation(ctx *service.SessionContext, zkey *crypto.Z
 
         // get response from Revocation service
         var resp message.Message
-        if resp, err = service.RequestResponse(ctx, "gns", "Revocation", config.Cfg.Revocation.Endpoint, req); err != nil {
+        if resp, err = service.RequestResponse(ctx, "gns", "Revocation", config.Cfg.Revocation.Service.Socket, req); err != nil {
                 return
         }
 
@@ -195,7 +198,7 @@ func (s *Service) QueryKeyRevocation(ctx *service.SessionContext, zkey *crypto.Z
 }
 
 // RevokeKey revokes a key with given revocation data
-func (s *Service) RevokeKey(ctx *service.SessionContext, rd *revocation.RevData) (success bool, err error) {
+func (s *Service) RevokeKey(ctx context.Context, rd *revocation.RevData) (success bool, err error) {
         logger.Printf(logger.DBG, "[gns] RevokeKey(%s)...\n", rd.ZoneKeySig.ID())
 
         // assemble request
@@ -206,7 +209,7 @@ func (s *Service) RevokeKey(ctx *service.SessionContext, rd *revocation.RevData)
 
         // get response from Revocation service
         var resp message.Message
-        if resp, err = service.RequestResponse(ctx, "gns", "Revocation", config.Cfg.Revocation.Endpoint, req); err != nil {
+        if resp, err = service.RequestResponse(ctx, "gns", "Revocation", config.Cfg.Revocation.Service.Socket, req); err != nil {
                 return
         }
 
@@ -225,17 +228,17 @@ func (s *Service) RevokeKey(ctx *service.SessionContext, rd *revocation.RevData)
 //======================================================================
 
 // LookupNamecache returns a cached lookup (if available)
-func (s *Service) LookupNamecache(ctx *service.SessionContext, query *Query) (block *message.Block, err error) {
-        logger.Printf(logger.DBG, "[gns] LookupNamecache(%s)...\n", hex.EncodeToString(query.Key.Bits))
+func (s *Service) LookupNamecache(ctx context.Context, query *blocks.GNSQuery) (block *blocks.GNSBlock, err error) {
+        logger.Printf(logger.DBG, "[gns] LookupNamecache(%s)...\n", hex.EncodeToString(query.Key().Bits))
 
         // assemble Namecache request
-        req := message.NewNamecacheLookupMsg(query.Key)
+        req := message.NewNamecacheLookupMsg(query.Key())
         req.ID = uint32(util.NextID())
         block = nil
 
         // get response from Namecache service
         var resp message.Message
-        if resp, err = service.RequestResponse(ctx, "gns", "Namecache", config.Cfg.Namecache.Endpoint, req); err != nil {
+        if resp, err = service.RequestResponse(ctx, "gns", "Namecache", config.Cfg.Namecache.Service.Socket, req); err != nil {
                 return
         }
 
@@ -250,7 +253,7 @@ func (s *Service) LookupNamecache(ctx *service.SessionContext, query *Query) (bl
                         break
                 }
                 // check if block was found
-                if len(m.EncData) == 0 || util.IsNull(m.EncData) {
+                if len(m.EncData) == 0 || util.IsAll(m.EncData, 0) {
                         logger.Println(logger.DBG, "[gns] block not found in namecache")
                         break
                 }
@@ -262,21 +265,21 @@ func (s *Service) LookupNamecache(ctx *service.SessionContext, query *Query) (bl
                 }
 
                 // assemble GNSBlock from message
-                block = new(message.Block)
+                block = new(blocks.GNSBlock)
                 block.DerivedKeySig = m.DerivedKeySig
-                sb := new(message.SignedBlockData)
+                sb := new(blocks.SignedGNSBlockData)
                 sb.Purpose = new(crypto.SignaturePurpose)
                 sb.Purpose.Purpose = enums.SIG_GNS_RECORD_SIGN
                 sb.Purpose.Size = uint32(16 + len(m.EncData))
                 sb.Expire = m.Expire
-                sb.EncData = m.EncData
-                block.Block = sb
+                sb.Data = m.EncData
+                block.Body = sb
 
                 // verify and decrypt block
-                if err = block.Verify(query.Zone, query.Label); err != nil {
+                if err = query.Verify(block); err != nil {
                         break
                 }
-                if err = block.Decrypt(query.Zone, query.Label); err != nil {
+                if err = query.Decrypt(block); err != nil {
                         break
                 }
         default:
@@ -287,7 +290,7 @@ func (s *Service) LookupNamecache(ctx *service.SessionContext, query *Query) (bl
 }
 
 // StoreNamecache stores a lookup in the local namecache.
-func (s *Service) StoreNamecache(ctx *service.SessionContext, block *message.Block) (err error) {
+func (s *Service) StoreNamecache(ctx context.Context, query *blocks.GNSQuery, block *blocks.GNSBlock) (err error) {
         logger.Println(logger.DBG, "[gns] StoreNamecache()...")
 
         // assemble Namecache request
@@ -296,7 +299,7 @@ func (s *Service) StoreNamecache(ctx *service.SessionContext, block *message.Blo
 
         // get response from Namecache service
         var resp message.Message
-        if resp, err = service.RequestResponse(ctx, "gns", "Namecache", config.Cfg.Namecache.Endpoint, req); err != nil {
+        if resp, err = service.RequestResponse(ctx, "gns", "Namecache", config.Cfg.Namecache.Service.Socket, req); err != nil {
                 return
         }
 
@@ -327,13 +330,13 @@ func (s *Service) StoreNamecache(ctx *service.SessionContext, block *message.Blo
 //======================================================================
 
 // LookupDHT gets a GNS block from the DHT for the given query key.
-func (s *Service) LookupDHT(ctx *service.SessionContext, query *Query) (block *message.Block, err error) {
-        logger.Printf(logger.DBG, "[gns] LookupDHT(%s)...\n", hex.EncodeToString(query.Key.Bits))
+func (s *Service) LookupDHT(ctx context.Context, query blocks.Query) (block blocks.Block, err error) {
+        logger.Printf(logger.DBG, "[gns] LookupDHT(%s)...\n", hex.EncodeToString(query.Key().Bits))
         block = nil
 
         // client-connect to the DHT service
         logger.Println(logger.DBG, "[gns] Connecting to DHT service...")
-        cl, err := service.NewClient(config.Cfg.DHT.Endpoint)
+        cl, err := service.NewClient(ctx, config.Cfg.DHT.Service.Socket)
         if err != nil {
                 return nil, err
         }
@@ -360,7 +363,7 @@ func (s *Service) LookupDHT(ctx *service.SessionContext, query *Query) (block *m
         )
 
         // send DHT GET request and wait for response
-        reqGet := message.NewDHTClientGetMsg(query.Key)
+        reqGet := message.NewDHTClientGetMsg(query.Key())
         reqGet.ID = uint64(util.NextID())
         reqGet.ReplLevel = uint32(enums.DHT_GNS_REPLICATION_LEVEL)
         reqGet.Type = uint32(enums.BLOCK_TYPE_GNS_NAMERECORD)
@@ -368,16 +371,16 @@ func (s *Service) LookupDHT(ctx *service.SessionContext, query *Query) (block *m
 
         if err = interact(reqGet, true); err != nil {
                 // check for aborted remote lookup: we need to cancel the query
-                if err == transport.ErrChannelInterrupted {
+                if err == service.ErrConnectionInterrupted {
                         logger.Println(logger.WARN, "[gns] remote Lookup aborted -- cleaning up.")
 
                         // send DHT GET_STOP request and terminate
-                        reqStop := message.NewDHTClientGetStopMsg(query.Key)
+                        reqStop := message.NewDHTClientGetStopMsg(query.Key())
                         reqStop.ID = reqGet.ID
                         if err = interact(reqStop, false); err != nil {
                                 logger.Printf(logger.ERROR, "[gns] remote Lookup abort failed: %s\n", err.Error())
                         }
-                        return nil, transport.ErrChannelInterrupted
+                        return nil, service.ErrConnectionInterrupted
                 }
         }
 
@@ -407,22 +410,24 @@ func (s *Service) LookupDHT(ctx *service.SessionContext, query *Query) (block *m
                 }
 
                 // get GNSBlock from message
-                block = message.NewBlock()
+                qGNS := query.(*blocks.GNSQuery)
+                block = new(blocks.GNSBlock)
                 if err = data.Unmarshal(block, m.Data); err != nil {
                         logger.Printf(logger.ERROR, "[gns] can't read GNS block: %s\n", err.Error())
                         break
                 }
+
                 // verify and decrypt block
-                if err = block.Verify(query.Zone, query.Label); err != nil {
+                if err = qGNS.Verify(block); err != nil {
                         break
                 }
-                if err = block.Decrypt(query.Zone, query.Label); err != nil {
+                if err = qGNS.Decrypt(block); err != nil {
                         break
                 }
 
                 // we got a result from DHT that was not in the namecache,
                 // so store it there now.
-                if err = s.StoreNamecache(ctx, block); err != nil {
+                if err = s.StoreNamecache(ctx, qGNS, block.(*blocks.GNSBlock)); err != nil {
                         logger.Printf(logger.ERROR, "[gns] can't store block in Namecache: %s\n", err.Error())
                 }
         }