Combining several goroutines with channels

I have the following use case:

1. "fetch" goroutine will fetch available data from db based on some predefined criteria.
2. then I will have, let's say 2 goroutines (process1, process2), each one of them is making some manipulation to the data (and the order is important).
3. the last goroutine (processSave) should update data in DB.

I understand that I need to use channels connecting each one of the goroutines with the next one in raw:
FetchToP1Chnl, P1ToP2Chnl, P2ToP3Chnl, P3ToSaveChnl.

Operations on the "working" object have to run in a sequential manner: fetch -> process1 -> process2 -> processSave .

Update:
I forgot to mention that process1 and process2 functions are changing the object. Process2 needs to work with the result of process1.

Questions that I'm not sure about:

• What kind of a channel is more suitable here: unbuffered or buffered (it the buffered one, so how to choose the optimal size)
• where is better to open those channels? (I believe that it should be done in main)
• where is better to place the closing for the channels? My application is expected to be running non-stop

I was thinking about the following approach:

type Object struct {
ID           string `bson:"_id"`
Data         string `bson:"data"`
Subdocument1 string // Added by process1
Subdocument2 string // Added by process2
}
func main() {
// Set up MongoDB client
clientOptions := options.Client().ApplyURI("mongodb://localhost:27017")
client, err := mongo.Connect(context.Background(), clientOptions)
if err != nil {
log.Fatal(err)
}
// Fetch objects from MongoDB
collection := client.Database("your-database").Collection("your-collection")
cursor, err := collection.Find(context.Background(), bson.M{})
if err != nil {
log.Fatal(err)
}
// Create channels for communication between goroutines
objectCh1 := make(chan Object)
objectCh2 := make(chan Object)
// Create wait group to wait for goroutines to finish
var wg sync.WaitGroup
// Start the fetcher goroutine
wg.Add(1)
go fetchObjects(cursor, objectCh1, &wg)
// Start process1 goroutine
wg.Add(1)
go process1(objectCh1, objectCh2, &wg)
// Start process2 goroutine
wg.Add(1)
go process2(objectCh2, &wg)
// Wait for all goroutines to finish
wg.Wait()
// Close MongoDB client
if err := client.Disconnect(context.Background()); err != nil {
log.Fatal(err)
}
fmt.Println("Done")
}
// fetchObjects fetches objects from the cursor and sends them to objectCh1
func fetchObjects(cursor *mongo.Cursor, objectCh1 chan<- Object, wg *sync.WaitGroup) {
defer close(objectCh1)
defer wg.Done()
for cursor.Next(context.Background()) {
var obj Object
err := cursor.Decode(&obj)
if err != nil {
log.Println(err)
continue
}
objectCh1 <- obj
}
if err := cursor.Err(); err != nil {
log.Println(err)
}
}
func process1(objectCh1 <-chan Object, objectCh2 chan<- Object, wg *sync.WaitGroup) {
defer wg.Done()
for obj := range objectCh1 {
obj.Subdocument1 = "subdocument1"
// Do additional processing or update in MongoDB
updatedObject:=updateObjectInMongoDB(obj)
objectCh2 <- updatedObject
}
close(objectCh2)
}
func process2(objectCh2 <-chan Object, wg *sync.WaitGroup) {
defer wg.Done()
for obj := range objectCh2 {
obj.Subdocument2 = "subdocument2"
// Do additional processing or update in MongoDB
updateObjectInMongoDB(obj)
}
}
// updateObjectInMongoDB is a placeholder function to update the object in MongoDB
func updateObjectInMongoDB(obj Object) Object {
fmt.Printf("Updated Object: %+v\n", obj)
// Your logic to update the object in MongoDB
}

Channels: you probably want buffered ones, so as to make it possible for sender goroutines to continue working even when reader ones are temporarily busy (I'm assuming here that you want to increase concurrency).

Where to open channels: it is good practice to "wire" the goroutines/channels graph outside the goroutines themselves, so yes, probably in main.

Closing channels: if your application is going non stop, then you may not have to close channels at all. But this is tricky - most serving applications don't really run forever. You may want to implement a graceful shutdown (e.g. in response to a signal). The rule of thumb is that it's the sender responsibility to close channels.

One additional point: it is not obvious why you need to design concurrency this way. You could have one fetcher goroutine (I assume the DB reads should not be parallelized) and then split the objects between several worker goroutines, each doing process1, process2 & save sequentially on a given object.