package quotes
import (
"context"
"encoding/json"
"fmt"
"net/http"
"time"
)
type Quote struct {
Vendor string
Price float64
Err error
}
type PriceClient struct {
HTTP *http.Client
BaseURL string
}
func NewPriceClient(base string) *PriceClient {
return &PriceClient{
HTTP: &http.Client{Timeout: 5 * time.Second},
BaseURL: base,
}
}
func (c *PriceClient) Fetch(ctx context.Context, vendor, symbol string) Quote {
url := fmt.Sprintf("%s/%s/quote?symbol=%s", c.BaseURL, vendor, symbol)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return Quote{Vendor: vendor, Err: err}
}
resp, err := c.HTTP.Do(req)
if err != nil {
return Quote{Vendor: vendor, Err: err}
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return Quote{Vendor: vendor, Err: fmt.Errorf("%s: status %d", vendor, resp.StatusCode)}
}
var body struct {
Price float64 `json:"price"`
}
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
return Quote{Vendor: vendor, Err: err}
}
return Quote{Vendor: vendor, Price: body.Price}
}
package quotes
import (
"context"
"sync"
)
const maxConcurrent = 4
func FetchAll(ctx context.Context, c *PriceClient, symbol string, vendors []string) ([]Quote, error) {
results := make(chan Quote)
sem := make(chan struct{}, maxConcurrent)
var wg sync.WaitGroup
for _, vendor := range vendors {
wg.Add(1)
go func(v string) {
defer wg.Done()
sem <- struct{}{} // acquire a slot
defer func() { <-sem }() // release it
select {
case results <- c.Fetch(ctx, v, symbol):
case <-ctx.Done():
}
}(vendor)
}
go func() {
wg.Wait()
close(results)
}()
collected := make([]Quote, 0, len(vendors))
for {
select {
case q, ok := <-results:
if !ok {
return collected, nil
}
collected = append(collected, q)
case <-ctx.Done():
return collected, ctx.Err()
}
}
}
package main
import (
"context"
"log"
"time"
"example.com/app/quotes"
)
func main() {
client := quotes.NewPriceClient("https://prices.internal")
vendors := []string{"acme", "globex", "initech", "umbrella", "stark"}
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
results, err := quotes.FetchAll(ctx, client, "AAPL", vendors)
if err != nil {
log.Printf("batch ended early: %v (got %d partial quotes)", err, len(results))
}
for _, q := range results {
if q.Err != nil {
log.Printf("%-8s failed: %v", q.Vendor, q.Err)
continue
}
log.Printf("%-8s price: %.2f", q.Vendor, q.Price)
}
}
This snippet demonstrates the fan-in concurrency pattern in Go: several API calls run in parallel and their results are collected on a single channel by a select loop that also watches for cancellation. It is split across a small client wrapper, a bounded fan-in coordinator, and the calling code that drives it.
In client.go, PriceClient wraps an *http.Client and exposes Fetch, a single-vendor request that honors the passed context.Context. The context is threaded into the request via http.NewRequestWithContext, so when the coordinator cancels, in-flight HTTP calls are torn down instead of leaking. Each result is packaged into a Quote value that carries the vendor name and any per-call error, which is the key to fan-in: failures travel on the same channel as successes rather than aborting the whole batch.
In fanin.go, FetchAll launches one goroutine per vendor but throttles concurrency with a buffered sem channel acting as a counting semaphore — each worker acquires a slot before calling and releases it after. Every worker sends its Quote onto the shared results channel. A separate goroutine closes results once sync.WaitGroup.Wait returns, which is what lets the consumer loop terminate cleanly. The select loop is the heart of the pattern: it reads from results until the channel is drained, but simultaneously watches ctx.Done() so a deadline or upstream cancellation exits promptly with ctx.Err(). The sem size caps outbound load, an important trade-off when calling flaky third parties.
In main.go, context.WithTimeout bounds the entire batch, and FetchAll returns partial results even when some vendors error. The caller inspects each Quote.Err individually, so one slow or failing vendor never sinks the others.
The pattern shines when independent I/O calls can proceed in parallel and the caller wants aggregate results with graceful degradation. Pitfalls to watch: always ensure the channel is closed exactly once, avoid sending on a closed channel after cancellation, and remember that a select on ctx.Done() may abandon still-running goroutines, so the context must actually reach them to prevent leaks.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
