Works, but doesn't work. Still need link ETSI to kms

kms/kms.go

@@ -33,6 +33,7 @@ type eKMS struct {
 	kmsUUID         uuid.UUID
 	qleMapMutex     sync.Mutex
 	QuantumElements map[qlElementId]*QuantumElement
+	kmsPeersMutex   sync.Mutex
 	KmsPeers        map[string]*kmsPeer
 	pbETSI.UnimplementedKmsETSIServer
 	pbIC.UnimplementedKmsTalkerServer
@@ -54,6 +55,7 @@ func NeweKMS(kmsName string, kmsUUID uuid.UUID) (newekms *eKMS) {
 		kmsName:         kmsName,
 		kmsUUID:         kmsUUID,
 		QuantumElements: make(map[qlElementId]*QuantumElement),
+		KmsPeers:        make(map[string]*kmsPeer),
 	}
 }
 
@@ -141,3 +143,24 @@ func (kms *eKMS) GenerateNewQleID() (qlElementId, error) {
 		//keep going....
 	}
 }
+
+// TODO/XXX error handling
+func (kms *eKMS) AddPeer(kmsPeerSocket string) {
+	//check if peer exists
+	if _, there := kms.KmsPeers[kmsPeerSocket]; there {
+		log.Fatalf("Trying to add existing peer %s", kmsPeerSocket)
+	}
+	peer := NewKmsPeer()
+	peer.tcpSocketStr = kmsPeerSocket
+
+	kms.kmsPeersMutex.Lock()
+	kms.KmsPeers[kmsPeerSocket] = &peer
+	kms.kmsPeersMutex.Unlock()
+
+	go peer.PeerHandler(kms.kmsName)
+}
+
+// TODO/XXX error handling
+func (kms *eKMS) RemovePeer(kmsPeerSocket string) {
+
+}

kms/kmsetsi.go

@@ -3,7 +3,6 @@ package kms
 import (
 	"context"
 	"flag"
-	"fmt"
 	"log"
 	"net"
 
@@ -15,7 +14,11 @@ var (
 	etsiPort = flag.Int("port", 50900, "The server port")
 )
 
-func (s *eKMS) ETSICapabilities(ctx context.Context, in *pb.ETSICapabilitiesRequest) (capReply *pb.ETSICapabilitiesReply, err error) {
+type etsiServer struct {
+	pb.UnimplementedKmsETSIServer
+}
+
+func (kms *etsiServer) ETSICapabilities(ctx context.Context, in *pb.ETSICapabilitiesRequest) (capReply *pb.ETSICapabilitiesReply, err error) {
 	log.Printf("Received: %v", in.GetMyKmsName())
 
 	return &pb.ETSICapabilitiesReply{
@@ -23,24 +26,26 @@ func (s *eKMS) ETSICapabilities(ctx context.Context, in *pb.ETSICapabilitiesRequ
 	}, nil
 }
 
-func (s *eKMS) ETSIAddKMSPeer(ctx context.Context, in *pb.ETSIKMSPeerRequest) (*pb.ETSIKMSPeerReply, error) {
+func (kms *etsiServer) ETSIAddKMSPeer(ctx context.Context, in *pb.ETSIKMSPeerRequest) (*pb.ETSIKMSPeerReply, error) {
+	//determine the kms structure to call
+	log.Printf("AddKMSPeer called.")
 
-	s.AddPeer(in.GetKmsPeerSocket())
+	//etsiServer.AddPeer(in.GetKmsPeerSocket())
 
 	return &pb.ETSIKMSPeerReply{
-		KmsPeerName: s.kmsName,
+		//KmsPeerName: kms.kmsName,
 	}, nil
 }
 
-func (s *eKMS) ETSIRemoveKMSPeer(ctx context.Context, in *pb.ETSIKMSPeerRequest) (*pb.ETSIKMSPeerReply, error) {
-	s.RemovePeer(in.GetKmsPeerSocket())
+func (kms *etsiServer) ETSIRemoveKMSPeer(ctx context.Context, in *pb.ETSIKMSPeerRequest) (*pb.ETSIKMSPeerReply, error) {
+	//kms.RemovePeer(in.GetKmsPeerSocket())
 
 	return &pb.ETSIKMSPeerReply{
-		KmsPeerName: s.kmsName,
+		//KmsPeerName: kms.kmsName,
 	}, nil
 }
 
-func (s *eKMS) GetEncryptKeys256Bit(ctx context.Context, in *pb.ETSIGetEncryptKeys256BitRequest) (*pb.ETSIGetEncryptKeys256BitReply, error) {
+func (kms *etsiServer) GetEncryptKeys256Bit(ctx context.Context, in *pb.ETSIGetEncryptKeys256BitRequest) (*pb.ETSIGetEncryptKeys256BitReply, error) {
 	log.Printf("Received request for n=%d keys", in.GetAmount())
 
 	testBytes := []byte{120, 120, 120}
@@ -52,15 +57,16 @@ func (s *eKMS) GetEncryptKeys256Bit(ctx context.Context, in *pb.ETSIGetEncryptKe
 	}, nil
 }
 
-func StartETSI() {
+func StartETSI(listenAddr string) {
 	flag.Parse()
 
-	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *etsiPort))
+	//lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *etsiPort))
+	lis, err := net.Listen("tcp", listenAddr)
 	if err != nil {
 		log.Fatalf("failed to listen: %v", err)
 	}
 	s := grpc.NewServer()
-	pb.RegisterKmsETSIServer(s, &eKMS{})
+	pb.RegisterKmsETSIServer(s, &etsiServer{})
 	log.Printf("server listening at %v", lis.Addr())
 	if err := s.Serve(lis); err != nil {
 		log.Fatalf("failed to serve: %v", err)

kms/kmsintercom.go

@@ -11,7 +11,11 @@ import (
 	"google.golang.org/grpc"
 )
 
-func (s *eKMS) InterComCapabilities(ctx context.Context, in *pb.InterComCapabilitiesRequest) (capReply *pb.InterComCapabilitiesReply, err error) {
+type kmsTalkerServer struct {
+	pb.UnimplementedKmsTalkerServer
+}
+
+func (s *kmsTalkerServer) InterComCapabilities(ctx context.Context, in *pb.InterComCapabilitiesRequest) (capReply *pb.InterComCapabilitiesReply, err error) {
 	log.Printf("Received: %v", in.GetMyKmsName())
 
 	return &pb.InterComCapabilitiesReply{
@@ -27,7 +31,7 @@ func StartInterComm(interComPort int) {
 		log.Fatalf("failed to listen: %v", err)
 	}
 	s := grpc.NewServer()
-	pb.RegisterKmsTalkerServer(s, &eKMS{})
+	pb.RegisterKmsTalkerServer(s, &kmsTalkerServer{})
 	log.Printf("server listening at %v", lis.Addr())
 	if err := s.Serve(lis); err != nil {
 		log.Fatalf("failed to serve: %v", err)

kms/kmspeers.go

@@ -12,33 +12,43 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 )
 
+type kmsPeerStatus int16
+
+const (
+	kmsPeerUp      kmsPeerStatus = iota // peer was seen up lately
+	kmsPeerDown                         // peer not reachable anymore
+	kmsPeerPending                      // peer was added and has to be succesfully contacted once
+	kmsPeerUnknown                      // not known, not initialized
+)
+
 type kmsPeerInfo interface {
 }
 
 type kmsPeer struct {
+	peerStatus   kmsPeerStatus
 	tcpSocket    net.TCPAddr // the IP address and TCP port (aka socket) of the kms peer
 	tcpSocketStr string      // string rep. of tcpSocket
 	name         string      // the name of the kms peer
 	id           uuid.UUID   // uuid of the peer
 }
 
-func NewKmsPeer() (peer *kmsPeer) {
-	return &kmsPeer{}
+func NewKmsPeer() (peer kmsPeer) {
+	return kmsPeer{
+		peerStatus: kmsPeerUnknown,
+	}
 }
 
-// TODO/XXX error handling
-func (kms *eKMS) AddPeer(kmsPeerSocket string) {
-	//check if peer exists
-	if _, there := kms.KmsPeers[kmsPeerSocket]; there {
-		log.Fatalf("Trying to add existing peer %s", kmsPeerSocket)
-	}
-	peer := NewKmsPeer()
-	peer.tcpSocketStr = kmsPeerSocket
+// Handles everything with respect to a specific KMS peer
+func (ph *kmsPeer) PeerHandler(kmsName string) {
+
+	log.Printf("%s started PeerHandler for %s:", kmsName, ph.tcpSocketStr)
 
 	// contact peer
-	newPeerConn, err := grpc.Dial(kmsPeerSocket, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	newPeerConn, err := grpc.Dial(ph.tcpSocketStr, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
-		log.Fatalf("did not connect: %v", err)
+		log.Printf("did not connect: %v", err)
+		ph.peerStatus = kmsPeerDown
+		return
 	}
 	defer newPeerConn.Close()
 
@@ -47,15 +57,12 @@ func (kms *eKMS) AddPeer(kmsPeerSocket string) {
 	// Contact the server and print out its response.
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	r, err := c.ETSICapabilities(ctx, &pb.ETSICapabilitiesRequest{MyKmsName: kms.kmsName})
+	r, err := c.ETSICapabilities(ctx, &pb.ETSICapabilitiesRequest{MyKmsName: kmsName})
 	if err != nil {
-		log.Fatalf("could not greet: %v", err)
+		log.Printf("could not greet: %v", err)
+		ph.peerStatus = kmsPeerDown
+		return
 	}
 	log.Printf("Greeting: %s", r.GetPeerKmsName())
 
 }
-
-// TODO/XXX error handling
-func (kms *eKMS) RemovePeer(kmsPeerSocket string) {
-
-}

main.go

@@ -46,9 +46,6 @@ func main() {
 
 	log.Println("Welcome to the proto-kms called: ", ql1Name)
 
-	// Start the SDN/management and key retrieval interface
-	go kms.StartETSI()
-
 	if selfTesting == true {
 		log.Printf("%s in self-testing mode", ql1Name)
 		go emulatedKMS(ql2Name, udpQL2AddrString, udpQL1AddrString)
@@ -76,6 +73,9 @@ func emulatedKMS(myName string, myUDPAddr string, peerUDPAddr string) {
 
 	myQL.QuantumElementLink.AddPeer(*udpQL2Addr)
 
+	// Start the SDN/management and key retrieval interface
+	go kms.StartETSI(peerUDPAddr)
+
 	// TODO/XXX catch errors!
 	emuKMS.GlobalKeyHandler(7 * time.Second)
 }