Go by examples

backoff.go

// Backoff is a simple backoff implementation.
type Backoff struct {
	MaxAttempts int
	Backoff     func(int) time.Duration
	attempts    int
	err         error
	nextCh      chan struct{}
}

var ErrMaxAttemptsReached = errors.New("max attempts reached")

func (r *Backoff) Next() <-chan struct{} {
	if r.nextCh == nil {
		r.attempts = 0
		r.nextCh = make(chan struct{}, 1)
	}

	r.attempts++

	if r.MaxAttempts > 0 && r.attempts > r.MaxAttempts {
		r.done(ErrMaxAttemptsReached)
		return r.nextCh
	}

	backoff := r.Backoff
	if backoff == nil {
		backoff = func(n int) time.Duration {
			d := n * int(time.Second)
			return time.Duration(d)
		}
	}

	time.AfterFunc(backoff(r.attempts), func() {
		r.nextCh <- struct{}{}
	})

	return r.nextCh
}

func (r *Backoff) Err() error {
	return r.err
}

var closedCh = make(chan struct{})

func init() {
	close(closedCh)
}

func (r *Backoff) done(err error) {
	if r.nextCh == nil {
		r.nextCh = closedCh
	} else {
		close(r.nextCh)
	}
	r.err = err
}

backoff_test.go

func ExampleBackoff() {
	backoff := Backoff{
		MaxAttempts: 3
	}

	ctx, cancel := context.WithCancel(context.Background())

	for {
		select {
		case <-backoff.Next():
			if err := backoff.Err(); err != nil {
				fmt.Printf("error: %v\n", err)
				return
			}
			fmt.Println("next attempt")
		case <-ctx.Done():
			fmt.Printf("canceled: %v\n", ctx.Err())
			return
		case <-time.After(5 * time.Second):
			fmt.Println("timed out")
			cancel()
		}
	}

	// Output:
}

throttling.go

package main

import (
	"context"
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"math/rand"
	"net/http"
	"sync"
	"time"
)

const httpPort = ":8000"

func init() {
	rand.Seed(int64(time.Now().Nanosecond()))
}

func main() {
	var wg sync.WaitGroup
	wg.Add(2)

	go func() {
		serverMain()
		wg.Done()
	}()

	go func() {
		for range time.Tick(300 * time.Millisecond) {
			err := clientMain()
			log.Println("client: error", err)
		}
	}()

	wg.Wait()
}

func clientMain() error {
	t := time.Now()
	resp, err := http.Get("http://localhost" + httpPort)
	log.Printf("client: request took %s", time.Since(t))
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		body, _ := ioutil.ReadAll(resp.Body)
		return fmt.Errorf("unexpected response %s: body %s", resp.Status, body)
	}
	return nil
}

func serverMain() error {
	log.Println("listening", httpPort)
	http.Handle("/", newServer())
	return http.ListenAndServe(httpPort, nil)
}

type server struct {
	svc *svc
}

func newServer() *server {
	return &server{
		svc: newSvc(),
	}
}

func (s *server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	err := s.svc.HandleRequest(req)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
	} else {
		fmt.Fprint(w, "Ok")
	}
}

type svc struct {
	rate  time.Duration
	reqCh chan *request
}

func newSvc() *svc {
	s := &svc{
		rate:  time.Second,
		reqCh: make(chan *request, 10),
	}
	go s.limitHandler()
	return s
}

func (s *svc) limitHandler() {
	limitter := time.NewTicker(s.rate)
	for req := range s.reqCh {
		<-limitter.C
		s.handleRequest(req)
	}
}

func (s *svc) HandleRequest(req *http.Request) error {
	log.Printf("svc: handling request")

	c := make(chan error)
	defer close(c)

	ctx, cancel := context.WithCancel(req.Context())
	defer cancel()

	s.reqCh <- &request{req: req, ctx: ctx, respCh: c}

	timer := time.NewTimer(2 * time.Second)
	defer timer.Stop()

	select {
	case err := <-c:
		return err
	case <-timer.C:
		cancel()
		return errors.New("request timeout")
	}

	return nil
}

func (s *svc) handleRequest(req *request) {
	var err error

	chance := rand.Intn(10)
	if chance > 6 {
		time.Sleep(4 * time.Second)
	}

	randErr := rand.Intn(10)
	if randErr > 8 {
		err = errors.New("random error")
	}

	select {
	case <-req.ctx.Done():
		return
	default:
		req.respCh <- err
	}
}

type request struct {
	req    *http.Request
	ctx    context.Context
	respCh chan<- error
}