multiple-producer-single-consumer-flow

main.go

@@ -66,10 +66,36 @@ func validateOrders(receivedOrdersChannel <-chan model.Order) (<-chan model.Orde
 func reserverInventory(in <-chan model.Order) <-chan model.Order {
 
 	out := make(chan model.Order)
+	workers := 3
+	var wg sync.WaitGroup
+	wg.Add(workers)
+
+	for i := 0; i < workers; i++ {
+		go func() {
+			defer wg.Done()
+
+			for order := range in {
+				order.Status = model.Reserved
+				out <- order
+			}
+			// close(out) multiple goroutiners trying to close the same channel
+		}()
+	}
+
+	go func() {
+		wg.Wait()
+		close(out)
+	}()
+
+	return out
+}
+
+func fillOrders(in <-chan model.Order) <-chan model.Order {
+	out := make(chan model.Order)
 
 	go func() {
 		for order := range in {
-			order.Status = model.Reserved
+			order.Status = model.Filled
 			out <- order
 		}
 		close(out)
@@ -83,17 +109,18 @@ func main() {
 	receivedOrdersChannel := receiveOrders()
 	validOrdersChannel, inValidOrdersChannel := validateOrders(receivedOrdersChannel)
 	reservedOrdersChannel := reserverInventory(validOrdersChannel)
+	filledOrdersChannel := fillOrders(reservedOrdersChannel)
 
 	var wg sync.WaitGroup
 	wg.Add(2)
 
-	go func(reservedOrderChannel <-chan model.Order) {
+	go func(filledOrdersChannel <-chan model.Order) {
 		defer wg.Done()
 
-		for reservedOrder := range reservedOrdersChannel {
-			fmt.Printf("Inventory recived for order -> %v \n", reservedOrder.String())
+		for filledOrder := range filledOrdersChannel {
+			fmt.Printf("Order has been completed -> %v \n", filledOrder.String())
 		}
-	}(receivedOrdersChannel)
+	}(filledOrdersChannel)
 
 	go func(invalidOrdersChannel <-chan model.InvalidOrder) {
 		defer wg.Done()