Merge pull request #2 from Famcache/fix/peer-mem

Memory Leak Fix

VERSION

@@ -1 +1 @@
-1.2.0
+1.2.1

pkg/server/handler.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"famcache/pkg/server/command"
 	"famcache/pkg/server/peers"
+	"io"
 )
 
 func (s *Server) handle(peer *peers.Peer) {
@@ -13,11 +14,15 @@ func (s *Server) handle(peer *peers.Peer) {
 		request, err := reader.ReadString('\n')
 
 		if err != nil {
+			if err == io.EOF {
+				s.peers.Remove(peer)
+				return
+			}
+
 			s.logger.Error("Error reading request")
 			return
 		}
 
-		println("Request: ", request)
 		com, err := command.NewCommand(request)
 
 		if err != nil {

pkg/server/impl.go

@@ -32,7 +32,7 @@ func (s *Server) Start() error {
 			return domain.ErrServerAccept
 		}
 
-		peer := s.AddPeer(&conn)
+		peer := s.peers.Add(conn)
 
 		s.logger.Info("New client connected: " + peer.ID())
 

pkg/server/peers/manager.go

@@ -0,0 +1,32 @@
+package peers
+
+import "net"
+
+type PeersManager struct {
+	peers map[string]Peer
+}
+
+func (pm *PeersManager) Add(conn net.Conn) *Peer {
+	peer := NewPeer(conn)
+	pm.peers[peer.ID()] = *peer
+
+	return peer
+}
+
+func (pm *PeersManager) Remove(peer *Peer) {
+	delete(pm.peers, peer.ID())
+}
+
+func (pm *PeersManager) All() *map[string]Peer {
+	return &pm.peers
+}
+
+func (pm *PeersManager) GetById(id string) Peer {
+	return pm.peers[id]
+}
+
+func NewPeersManager() PeersManager {
+	return PeersManager{
+		peers: make(map[string]Peer),
+	}
+}

pkg/server/publish_handler.go

@@ -9,7 +9,7 @@ import (
 func (s *Server) handlePublish(peer *peers.Peer, message *command.MessagingCommand) {
 	s.logger.Info("Peer " + peer.ID() + " published topic " + message.Topic)
 
-	for _, p := range s.peers {
+	for _, p := range *s.peers.All() {
 		if p.ID() == peer.ID() {
 			continue
 		}

pkg/server/server.go

@@ -16,7 +16,7 @@ type Server struct {
 	port           string
 	logger         logger.Logger
 	cache          cache.Cache
-	peers          []peers.Peer
+	peers          peers.PeersManager
 	messagingQueue pubsub.Queue
 }
 
@@ -26,7 +26,7 @@ func NewServer(options ServerOptions) server.FamcacheServer {
 		port:           options.Port,
 		logger:         options.Logger,
 		cache:          options.Cache,
-		peers:          make([]peers.Peer, 0),
+		peers:          peers.NewPeersManager(),
 		messagingQueue: queue.NewPubsubQueue(),
 	}
 }