batch_events.go

package main

import (
  "time"
)

type Batch struct {
  events []int
}

func writeInfiniteEvents() <-chan int {
	eventsChan := make(chan int)
  go func() {
    for {
      eventsChan <- 1
    }
  }()
  return eventsChan
}

func processBatch(b *Batch) {
  // do something
}

func main() {
  // dummy event writer
  events := writeInfiniteEvents()
  batchSize := 1000
  // this channel will have a batch
  batchOut := make(chan *Batch)

  go func() {
    // initialize empty batch
    eventCount := 0
    batch := &Batch{}
    for {
      select {

        // read new event
        case event := <-events:

        // append it to current batch
        batch.events = append(batch.events, event)
        eventCount++

        // check if we have reached the limit
        if len(batch.events) == batchSize {
          // pass it to batchOut
          batchOut <- batch

          // reset
          batch = &Batch{}
          eventCount = 0
        }

        // check if we reached 5 seconds
        case <-time.After(5 * time.Second):
        // pass it to batch out
        batchOut <- batch

        // reset
        batch = &Batch{}
        eventCount = 0
      }
    }
  }()

  go func() {
    // Spin up new routines as we get a new batch
    for b := range batchOut {
      go processBatch(b)
    }
  }()
  
}