channels.go

channels.go

func listenAndUpsert() {
   c := make(chan User)    // create a channel
   go listenFromKafka(c)  // pass channel to a goroutine

   for msg := range c {
      upsertToCouchbase(msg, "key")
   }
}

func listenFromKafka(c chan User) {   

   for {
      random := rand.Intn(100)       // simulating incoming message
      time.Sleep(2 * time.Millisecond) // simulating kafka with 2ms
      fmt.Println("# ID " + strconv.Itoa(random) +
                  " received from Kafka")
      c <- User{Id: random}
   }
}
func upsertToCouchbase(user User, key string) {
  
   fmt.Println("# upserting to Couchbase for " +
   strconv.Itoa(user.Id))
   time.Sleep(5 * time.Millisecond)
}