pdf 14 done

src/banktransfer/main.go

@@ -32,6 +32,9 @@ func main() {
 		log.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
 	}
 	grpcServer := grpc.NewServer()
+	transferService := service.NewBankTransferService()
+	transferService.Start()
+	defer transferService.Stop()
 	banktransfer.RegisterBankTransferServer(grpcServer, service.NewBankTransferService())
 	if err := grpcServer.Serve(lis); err != nil {
 		log.Fatalf("failed to serve: %v", err)

src/banktransfer/service/banktransfer.go

@@ -11,15 +11,23 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
+const retryTime = 5 * time.Second
+
 type BankTransferService struct {
 	banktransfer.BankTransferServer
-	counter int32
-	queue   chan banktransfer.Transaction
+	counter    int32
+	queue      chan banktransfer.Transaction
+	retryQueue chan banktransfer.Transaction
+	stop       chan struct{}
 }
 
 func NewBankTransferService() *BankTransferService {
 	rand.Seed(time.Now().UnixNano())
-	return &BankTransferService{counter: 1, queue: make(chan banktransfer.Transaction)}
+	return &BankTransferService{counter: 0,
+		queue:      make(chan banktransfer.Transaction),
+		retryQueue: make(chan banktransfer.Transaction),
+		stop:       make(chan struct{}),
+	}
 }
 func (s *BankTransferService) TransferMoney(_ context.Context, transaction *banktransfer.Transaction) (*emptypb.Empty, error) {
 	entry := log.WithField("transaction", transaction)
@@ -32,7 +40,7 @@ func (s *BankTransferService) processTransaction(transaction *banktransfer.Trans
 	entry := log.WithField("transaction", transaction)
 	go func(transaction banktransfer.Transaction) {
 		entry.Info("Start processing transaction")
-		atomic.AddInt32(&s.counter, 1)
+		transaction.Id = s.getUniqueId()
 		time.Sleep(time.Duration(rand.Intn(9)+1) * time.Second)
 		s.queue <- transaction
 		entry.Info("Processing transaction finished")
@@ -51,14 +59,14 @@ func (s *BankTransferService) ProcessTransactions(stream banktransfer.BankTransf
 				entry := log.WithField("transaction", transaction)
 				entry.Info("Sending transaction")
 				if err := stream.Send(&transaction); err != nil {
-					// TODO: we need to recover from this if communication is not up
+					s.requeueTransaction(&transaction)
 					entry.WithError(err).Error("Error sending transaction")
 					return err
 				}
 				entry.Info("Transaction sent. Waiting for processing response")
 				response, err := stream.Recv()
 				if err != nil {
-					// TODO: we need to recover from this if communication is not up
+					s.requeueTransaction(&transaction)
 					entry.WithError(err).Error("Error receiving processing response")
 					return err
 				}
@@ -72,3 +80,35 @@ func (s *BankTransferService) ProcessTransactions(stream banktransfer.BankTransf
 		}
 	}()
 }
+
+func (s *BankTransferService) requeueTransaction(transaction *banktransfer.Transaction) {
+	entry := log.WithField("transaction", transaction)
+	go func(transaction banktransfer.Transaction) {
+		entry.Infof("Requeuing transaction. Wait for %f seconds", retryTime.Seconds())
+		time.Sleep(retryTime)
+		s.retryQueue <- transaction
+		entry.Info("Requeued transaction")
+	}(*transaction)
+}
+
+func (s *BankTransferService) getUniqueId() int32 {
+	return atomic.AddInt32(&s.counter, 1)
+}
+func (s *BankTransferService) Start() {
+	log.Info("Starting banktransfer service")
+	go func() {
+		for {
+			select {
+			case <-s.stop:
+				break
+			case transaction := <-s.retryQueue:
+				s.queue <- transaction
+			}
+		}
+	}()
+}
+
+func (s *BankTransferService) Stop() {
+	log.Info("Stopping banktransfer service")
+	close(s.stop)
+}

src/myaktion/monitor.go

@@ -18,7 +18,7 @@ func monitortransactions() {
 }
 
 func connectandmonitor() {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
 	defer cancel()
 	conn, err := client.GetBankTransferConnection(ctx)
 	if err != nil {