go.mod
module github.com/example/concurrent-fetch
go 1.21
main.go
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"os"
"os/signal"
"path/filepath"
"sort"
"strings"
"syscall"
"time"
)
type URLResult struct {
URL string json:"url"
Bytes int json:"bytes"
Headings int json:"headings"
Attempts int json:"attempts"
Cached bool json:"cached"
DurationMS int64 json:"duration_ms"
Error string json:"error,omitempty"
}
type Totals struct {
URLs int json:"urls"
Successes int json:"successes"
Failures int json:"failures"
Bytes int json:"bytes"
Headings int json:"headings"
}
type Meta struct {
Workers int json:"workers"
Timeout string json:"timeout"
Retries int json:"retries"
Backoff string json:"backoff"
CacheDir string json:"cache_dir"
CacheTTL string json:"cache_ttl"
Started string json:"started"
Duration string json:"duration"
}
type Summary struct {
Results []URLResult json:"results"
Totals Totals json:"totals"
Meta Meta json:"meta"
}
// Example usage:
// go run ./cmd -workers=8 -timeout=4s -retries=2 -backoff=300ms https://example.com https://golang.org
// go run . -urls-file=urls.txt
func main() {
workers := flag.Int("workers", 4, "number of concurrent workers")
timeout := flag.Duration("timeout", 5time.Second, "per-request timeout")
retries := flag.Int("retries", 2, "retries on transient errors")
backoff := flag.Duration("backoff", 300time.Millisecond, "base backoff between retries")
cacheDir := flag.String("cache-dir", "", "cache directory (default: $TMP/concurrent-fetch-cache)")
cacheTTL := flag.Duration("cache-ttl", 10*time.Minute, "use cached content if not older than this duration")
urlsFile := flag.String("urls-file", "", "path to file with URLs (one per line); if empty, takes URLs from args")
logLevel := flag.String("log-level", "info", "log level: debug, info, warn, error")
flag.Parse()
urls := []string{}
if *urlsFile != "" {
b, err := os.ReadFile(*urlsFile)
if err != nil {
fmt.Fprintf(os.Stderr, "error reading urls file: %v\n", err)
os.Exit(1)
}
for _, l := range strings.Split(string(b), "\n") {
l = strings.TrimSpace(l)
if l != "" && !strings.HasPrefix(l, "#") {
urls = append(urls, l)
}
}
} else if flag.NArg() > 0 {
urls = flag.Args()
} else {
urls = []string{
"https://example.com",
"https://example.com/",
}
}
if len(urls) == 0 {
fmt.Fprintln(os.Stderr, "no URLs provided")
os.Exit(1)
}
if *cacheDir == "" {
*cacheDir = filepath.Join(os.TempDir(), "concurrent-fetch-cache")
}
logger := NewLogger(os.Stderr, ParseLevel(*logLevel))
logger.Info("starting", Fields{
"workers": *workers,
"timeout": timeout.String(),
"retries": *retries,
"backoff": backoff.String(),
"cache_dir": *cacheDir,
"cache_ttl": cacheTTL.String(),
"urls": len(urls),
})
if err := os.MkdirAll(*cacheDir, 0o755); err != nil {
logger.Error("failed to create cache dir", Fields{"err": err.Error(), "dir": *cacheDir})
os.Exit(1)
}
cache := NewFileCache(*cacheDir)
fetcher := &Fetcher{
Timeout: *timeout,
Retries: *retries,
Backoff: *backoff,
Cache: cache,
TTL: *cacheTTL,
Logger: logger,
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
sigCh := make(chan os.Signal, 2)
signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
go func() {
s := <-sigCh
logger.Warn("signal received, shutting down", Fields{"signal": s.String()})
cancel()
// Second signal forces exit
s = <-sigCh
logger.Error("second signal received, forcing exit", Fields{"signal": s.String()})
os.Exit(2)
}()
start := time.Now()
jobs := make(chan Job)
resultsCh := make(chan URLResult)
// Start workers
go func() {
RunWorkers(ctx, *workers, jobs, func(ctx context.Context, j Job) URLResult {
t0 := time.Now()
data, attempts, cached, err := fetcher.Fetch(ctx, j.URL)
res := URLResult{
URL: j.URL,
Attempts: attempts,
Cached: cached,
DurationMS: time.Since(t0).Milliseconds(),
}
if err != nil {
res.Error = err.Error()
return res
}
res.Bytes = len(data)
res.Headings = CountHeadings(data)
return res
}, resultsCh)
}()
// Feed jobs
go func() {
defer close(jobs)
for _, u := range urls {
select {
case <-ctx.Done():
return
case jobs <- Job{URL: strings.TrimSpace(u)}:
}
}
}()
// Collect results
collected := make([]URLResult, 0, len(urls))
for r := range resultsCh {
collected = append(collected, r)
if len(collected) == len(urls) {
break
}
}
// Ensure workers stop
cancel()
// Aggregate
sort.Slice(collected, func(i, j int) bool { return collected[i].URL < collected[j].URL })
var totals Totals
totals.URLs = len(collected)
for _, r := range collected {
if r.Error == "" {
totals.Successes++
totals.Bytes += r.Bytes
totals.Headings += r.Headings
} else {
totals.Failures++
}
}
sum := Summary{
Results: collected,
Totals: totals,
Meta: Meta{
Workers: *workers,
Timeout: timeout.String(),
Retries: *retries,
Backoff: backoff.String(),
CacheDir: *cacheDir,
CacheTTL: cacheTTL.String(),
Started: start.Format(time.RFC3339Nano),
Duration: time.Since(start).String(),
},
}
enc := json.NewEncoder(os.Stdout)
enc.SetIndent("", " ")
if err := enc.Encode(sum); err != nil {
logger.Error("failed to write summary", Fields{"err": err.Error()})
os.Exit(1)
}
logger.Info("done", Fields{"duration": time.Since(start).String(), "successes": totals.Successes, "failures": totals.Failures})
}
logger.go
package main
import (
"encoding/json"
"io"
"strings"
"sync"
"time"
)
type Level int
const (
LevelDebug Level = iota
LevelInfo
LevelWarn
LevelError
)
func ParseLevel(s string) Level {
switch strings.ToLower(strings.TrimSpace(s)) {
case "debug":
return LevelDebug
case "warn", "warning":
return LevelWarn
case "error":
return LevelError
default:
return LevelInfo
}
}
type Fields map[string]any
type Logger struct {
mu sync.Mutex
enc *json.Encoder
level Level
}
func NewLogger(w io.Writer, level Level) *Logger {
return &Logger{enc: json.NewEncoder(w), level: level}
}
func (l *Logger) log(level Level, msg string, fields Fields) {
if level < l.level {
return
}
entry := map[string]any{
"ts": time.Now().Format(time.RFC3339Nano),
"lvl": [...]string{"debug", "info", "warn", "error"}[level],
"msg": msg,
"pid": getPID(),
"prog": getProg(),
}
for k, v := range fields {
entry[k] = v
}
l.mu.Lock()
defer l.mu.Unlock()
_ = l.enc.Encode(entry)
}
func (l *Logger) Debug(msg string, f Fields) { l.log(LevelDebug, msg, f) }
func (l *Logger) Info(msg string, f Fields) { l.log(LevelInfo, msg, f) }
func (l *Logger) Warn(msg string, f Fields) { l.log(LevelWarn, msg, f) }
func (l *Logger) Error(msg string, f Fields) { l.log(LevelError, msg, f) }
func getPID() int {
return osGetpid()
}
func getProg() string {
return osArgs0()
}
logger_os.go
package main
import "os"
func osGetpid() int { return os.Getpid() }
func osArgs0() string {
if len(os.Args) > 0 {
return os.Args[0]
}
return "app"
}
worker.go
package main
import (
"context"
"sync"
)
type Job struct {
URL string
}
type Handler func(context.Context, Job) URLResult
func RunWorkers(ctx context.Context, n int, jobs <-chan Job, handler Handler, out chan<- URLResult) {
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func(id int) {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case j, ok := <-jobs:
if !ok {
return
}
out <- handler(ctx, j)
}
}
}(i)
}
go func() {
wg.Wait()
close(out)
}()
}
fetcher.go
package main
import (
"context"
"errors"
"fmt"
"io"
"math/rand"
"net"
"net/http"
"time"
)
type Cache interface {
Get(url string, ttl time.Duration) ([]byte, bool, error)
Put(url string, data []byte) error
}
type Fetcher struct {
Timeout time.Duration
Retries int
Backoff time.Duration
Cache Cache
TTL time.Duration
Logger *Logger
Client *http.Client // optional; if nil, constructed per request
}
func (f *Fetcher) httpClient() *http.Client {
if f.Client != nil {
return f.Client
}
return &http.Client{
Timeout: f.Timeout,
Transport: &http.Transport{
Proxy: http.ProxyFromEnvironment,
MaxIdleConns: 100,
MaxConnsPerHost: 20,
MaxIdleConnsPerHost: 20,
IdleConnTimeout: 90 * time.Second,
DialContext: (&net.Dialer{
Timeout: 5 * time.Second,
KeepAlive: 30 * time.Second,
}).DialContext,
},
}
}
func (f *Fetcher) Fetch(ctx context.Context, url string) ([]byte, int, bool, error) {
if f.Cache != nil && f.TTL > 0 {
if data, ok, err := f.Cache.Get(url, f.TTL); err == nil && ok {
f.Logger.Debug("cache hit", Fields{"url": url})
return data, 0, true, nil
}
}
attempts := 0
var lastErr error
for {
select {
case <-ctx.Done():
return nil, attempts, false, ctx.Err()
default:
}
attempts++
data, err := f.fetchOnce(ctx, url)
if err == nil {
if f.Cache != nil {
_ = f.Cache.Put(url, data)
}
return data, attempts, false, nil
}
lastErr = err
if attempts > f.Retries {
break
}
// backoff with jitter
delay := f.Backoff + time.Duration(rand.Int63n(int64(f.Backoff)/2+1))
if delay <= 0 {
delay = 100 * time.Millisecond
}
f.Logger.Warn("fetch failed, retrying", Fields{"url": url, "attempt": attempts, "err": err.Error(), "sleep": delay.String()})
t := time.NewTimer(delay)
select {
case <-ctx.Done():
t.Stop()
return nil, attempts, false, ctx.Err()
case <-t.C:
}
}
return nil, attempts, false, fmt.Errorf("fetch failed after %d attempt(s): %w", attempts, lastErr)
}
func (f *Fetcher) fetchOnce(ctx context.Context, url string) ([]byte, error) {
client := f.httpClient()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, err
}
req.Header.Set("User-Agent", "concurrent-fetch/1.0 (+https://example)")
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode >= 500 {
return nil, fmt.Errorf("server error: %d", resp.StatusCode)
}
if resp.StatusCode >= 400 {
// treat 4xx as non-retryable error
return nil, fmt.Errorf("client error: %d", resp.StatusCode)
}
limited := &io.LimitedReader{R: resp.Body, N: 10 << 20} // 10MB safety limit
data, err := io.ReadAll(limited)
if err != nil {
return nil, err
}
if limited.N == 0 {
return nil, errors.New("response too large")
}
return data, nil
}
parser.go
package main
import (
"bytes"
"strings"
"golang.org/x/net/html"
)
func CountHeadings(content []byte) int {
// Try HTML first
if n, ok := countHTMLHeadings(content); ok {
return n
}
// Fallback: Markdown-style headings
return countMarkdownHeadings(content)
}
func countHTMLHeadings(content []byte) (int, bool) {
doc, err := html.Parse(bytes.NewReader(content))
if err != nil {
return 0, false
}
count := 0
var walk func(*html.Node)
walk = func(n *html.Node) {
if n.Type == html.ElementNode {
switch strings.ToLower(n.Data) {
case "h1", "h2", "h3", "h4", "h5", "h6":
count++
}
}
for c := n.FirstChild; c != nil; c = c.NextSibling {
walk(c)
}
}
walk(doc)
return count, true
}
func countMarkdownHeadings(content []byte) int {
lines := strings.Split(string(content), "\n")
count := 0
for _, l := range lines {
lt := strings.TrimSpace(l)
if lt == "" || strings.HasPrefix(lt, "<") { // probably HTML; skip noisy matches
continue
}
if strings.HasPrefix(lt, "#") {
count++
}
}
return count
}
cache.go
package main
import (
"crypto/sha256"
"encoding/hex"
"errors"
"os"
"path/filepath"
"time"
)
type FileCache struct {
dir string
}
func NewFileCache(dir string) *FileCache {
return &FileCache{dir: dir}
}
func (c *FileCache) key(url string) string {
sum := sha256.Sum256([]byte(url))
return filepath.Join(c.dir, hex.EncodeToString(sum[:])+".cache")
}
func (c *FileCache) Get(url string, ttl time.Duration) ([]byte, bool, error) {
path := c.key(url)
st, err := os.Stat(path)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return nil, false, nil
}
return nil, false, err
}
if ttl > 0 && time.Since(st.ModTime()) > ttl {
return nil, false, nil
}
data, err := os.ReadFile(path)
if err != nil {
return nil, false, err
}
return data, true, nil
}
func (c *FileCache) Put(url string, data []byte) error {
path := c.key(url)
tmp := path + ".tmp"
if err := os.WriteFile(tmp, data, 0o644); err != nil {
return err
}
return os.Rename(tmp, path)
}
go.sum
// intentionally empty; no external dependencies recorded for stdlib only
parser_test.go
package main
import "testing"
func TestCountHeadingsHTML(t *testing.T) {
html := `
x
A
no
`
got := CountHeadings([]byte(html))
if got != 3 {
t.Fatalf("expected 3 headings, got %d", got)
}
}
func TestCountHeadingsMarkdown(t *testing.T) {
md := `
Title
some text
Subtitle
normal
Third
not a heading # text
`
got := CountHeadings([]byte(md))
if got != 3 {
t.Fatalf("expected 3 headings, got %d", got)
}
}
fetcher_test.go
package main
import (
"context"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"sync/atomic"
"testing"
"time"
)
func TestFetcherRetry(t *testing.T) {
var hits int32
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
n := atomic.AddInt32(&hits, 1)
if n <= 2 {
http.Error(w, "temporary", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/plain")
w.Write([]byte("
ok
"))
}))
defer srv.Close()
logger := NewLogger(os.Stderr, LevelError)
f := &Fetcher{
Timeout: 2 * time.Second,
Retries: 3,
Backoff: 10 * time.Millisecond,
TTL: 0,
Logger: logger,
Client: srv.Client(),
}
ctx := context.Background()
data, attempts, cached, err := f.Fetch(ctx, srv.URL)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if cached {
t.Fatalf("unexpected cached result")
}
if attempts < 3 {
t.Fatalf("expected at least 3 attempts, got %d", attempts)
}
if string(data) != "<h1>ok</h1>" {
t.Fatalf("unexpected body: %q", string(data))
}
}
func TestFetcherCache(t *testing.T) {
dir := t.TempDir()
cache := NewFileCache(dir)
logger := NewLogger(os.Stderr, LevelError)
f := &Fetcher{
Timeout: 1 * time.Second,
Retries: 0,
Backoff: 10 * time.Millisecond,
TTL: 10 * time.Minute,
Cache: cache,
Logger: logger,
Client: &http.Client{Timeout: 1 * time.Second},
}
url := "https://example.com/cached"
// Seed cache
data := []byte("hello # world")
if err := os.WriteFile(filepath.Join(dir, cache.key(url)), data, 0o644); err != nil {
t.Fatalf("write cache: %v", err)
}
got, attempts, cached, err := f.Fetch(context.Background(), url)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !cached {
t.Fatalf("expected cached")
}
if attempts != 0 {
t.Fatalf("expected 0 attempts when cached, got %d", attempts)
}
if string(got) != "hello # world" {
t.Fatalf("unexpected cached content: %q", string(got))
}
}
integration_test.go
package main
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"sort"
"testing"
"time"
)
func TestEndToEnd(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/a", func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte("
a
x
"))
})
mux.HandleFunc("/b", func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte("# md\n## md2\ntext"))
})
srv := httptest.NewServer(mux)
defer srv.Close()
urls := []string{srv.URL + "/a", srv.URL + "/b"}
logger := NewLogger(os.Stderr, LevelError)
cache := NewFileCache(t.TempDir())
fetcher := &Fetcher{
Timeout: 1 * time.Second,
Retries: 1,
Backoff: 10 * time.Millisecond,
TTL: 0,
Cache: cache,
Logger: logger,
Client: srv.Client(),
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
jobs := make(chan Job)
out := make(chan URLResult)
go func() {
RunWorkers(ctx, 2, jobs, func(ctx context.Context, j Job) URLResult {
t0 := time.Now()
data, attempts, cached, err := fetcher.Fetch(ctx, j.URL)
r := URLResult{URL: j.URL, Attempts: attempts, Cached: cached, DurationMS: time.Since(t0).Milliseconds()}
if err != nil {
r.Error = err.Error()
return r
}
r.Bytes = len(data)
r.Headings = CountHeadings(data)
return r
}, out)
}()
go func() {
defer close(jobs)
for _, u := range urls {
jobs <- Job{URL: u}
}
}()
var results []URLResult
for r := range out {
results = append(results, r)
if len(results) == len(urls) {
break
}
}
if len(results) != 2 {
t.Fatalf("expected 2 results, got %d", len(results))
}
sort.Slice(results, func(i, j int) bool { return results[i].URL < results[j].URL })
if results[0].Headings != 2 || results[1].Headings != 2 {
b, _ := json.MarshalIndent(results, "", " ")
t.Fatalf("unexpected headings count: %s", string(b))
}
}
go.mod note:
This code uses golang.org/x/net/html for robust HTML heading parsing. Add the following requirement if your environment needs an explicit module line:
- run: go get golang.org/x/net@latest
Example usage:
Behavior:
- Structured JSON logs to stderr.
- Final aggregated JSON printed to stdout.
- Graceful shutdown on Ctrl+C (first cancels in-flight work, second exits).
