package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
const (
bufferSize = 32
numProducers = 2
numConsumers = 2
messagesPerProducer = 10
)
type message struct {
content string
}
type printMessage struct {
source string
content string
}
func producer(id int, messages chan<- message, printer chan<- printMessage, wg *sync.WaitGroup) {
defer wg.Done()
for i := 0; i < messagesPerProducer; i++ {
msg := message{fmt.Sprintf("Message %d from Producer %d", i, id)}
messages <- msg
printer <- printMessage{"producer", fmt.Sprintf("Producer %d sent: %s", id, msg.content)}
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
}
messages <- message{"Bye"}
}
func consumer(id int, messages <-chan message, printer chan<- printMessage, wg *sync.WaitGroup) {
defer wg.Done()
for msg := range messages {
printer <- printMessage{"consumer", fmt.Sprintf("Consumer %d received: %s", id, msg.content)}
if msg.content == "Bye" {
return
}
time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
}
}
func printer(producerMsgs <-chan printMessage, consumerMsgs <-chan printMessage, done <-chan bool) {
const TIMEOUT_MS = 2000
for {
select {
case msg := <-producerMsgs:
fmt.Printf("[PRODUCER] %s\n", msg.content)
case msg := <-consumerMsgs:
fmt.Printf("[CONSUMER] %s\n", msg.content)
case <-done:
fmt.Println("Printer: Shutting down...")
return
case <-time.After(TIMEOUT_MS * time.Millisecond):
fmt.Println("Printer: No activity for", TIMEOUT_MS, "ms. Shutting down...")
}
}
}
func main() {
rand.Seed(time.Now().UnixNano())
messages := make(chan message, bufferSize)
producerMsgs := make(chan printMessage, bufferSize)
consumerMsgs := make(chan printMessage, bufferSize)
done := make(chan bool)
var wg sync.WaitGroup
// Start printer
go printer(producerMsgs, consumerMsgs, done)
// Start producers
for i := 0; i < numProducers; i++ {
wg.Add(1)
go producer(i, messages, producerMsgs, &wg)
}
// Start consumers
for i := 0; i < numConsumers; i++ {
wg.Add(1)
go consumer(i, messages, consumerMsgs, &wg)
}
// Wait for all goroutines to finish
wg.Wait()
close(messages)
// Signal printer to shut down
done <- true
// Allow some time for final messages to be printed
time.Sleep(100 * time.Millisecond)
}
