package main

import (
    "fmt"
    "log"
    "sync"
    "time"

    "github.com/streadway/amqp"
)

var queue []*amqp.Delivery
var waitGroup sync.WaitGroup

func bufferDeliveries(in chan *amqp.Delivery, out chan amqp.Delivery) {
    defer waitGroup.Done()
    var queueIn = in

    for delivery := range in {
        select {
        case out <- *delivery:
            // delivered immediately while the consumer chan can receive
        default:
            queue = append(queue, delivery)
        }

        for len(queue) > 0 {
            select {
            case out <- *queue[0]:
                queue = queue[1:]
            case delivery, open := <-queueIn:
                if open {
                    queue = append(queue, delivery)
                } else {
                    // stop receiving to drain the queue
                    queueIn = nil
                }
            }
        }
    }

    close(out)
}

func main() {
    in := make(chan *amqp.Delivery)
    out := make(chan amqp.Delivery)

    waitGroup.Add(3)
    go bufferDeliveries(in, out)

    go func() {
        defer waitGroup.Done()
        counter := 0
        for {
            counter++
            //log.Println("send", counter)
            in <- &amqp.Delivery{
                Body: []byte(fmt.Sprintf("%d", counter)),
            }
            if counter == 1000000 {
                close(in)
                return
            }
        }
    }()

    go func() {
        defer waitGroup.Done()
        pre := 0
        var counter int
        for d := range out {
            fmt.Sscanf(string(d.Body), "%d", &counter)
            if pre >= counter {
                log.Println("bad order", pre, counter, string(d.Body))
                panic("bad order")
            }
            pre = counter
            <-time.After(time.Microsecond)
        }
    }()
    waitGroup.Wait()
    log.Println("the last:", cap(queue), len(queue))
    <-time.After(time.Hour)
}