text:
# default client timeout to use with blocking methods after which they throw an error
# Julia Timer converts seconds to milliseconds and adds 1 to it before passing it to libuv, hence the magic numbers to prevent overflow
const DEFAULT_TIMEOUT = round(Int, typemax(Int)/1000) - 1
const DEFAULT_CONNECT_TIMEOUT = round(Int, typemax(Int)/1000) - 1

# ----------------------------------------
# IO for types begin
# ----------------------------------------
function read(io::IO, ::Type{TAMQPBit})
    TAMQPBit(ntoh(read(io, UInt8)))
end

function write(io::IO, b::TAMQPBit)
    write(io, hton(b.val))
end

function read(io::IO, ::Type{TAMQPFrameProperties})
    TAMQPFrameProperties(
        ntoh(read(io, fieldtype(TAMQPFrameProperties, :channel))),
        ntoh(read(io, fieldtype(TAMQPFrameProperties, :payloadsize))),
    )
end

write(io::IO, p::TAMQPFrameProperties) = write(io, hton(p.channel), hton(p.payloadsize))

function read!(io::IO, b::TAMQPBodyPayload)
    read!(io, b.data)
    b
end

write(io::IO, b::TAMQPBodyPayload) = write(io, b.data)

function read(io::IO, ::Type{TAMQPShortStr})
    len = ntoh(read(io, TAMQPOctet))
    TAMQPShortStr(len, read!(io, Vector{UInt8}(undef, len)))
end

function read(io::IO, ::Type{TAMQPLongStr})
    len = ntoh(read(io, TAMQPLongUInt))
    TAMQPLongStr(len, read!(io, Vector{UInt8}(undef, len)))
end

function read(io::IO, ::Type{TAMQPByteArray})
    len = ntoh(read(io, TAMQPLongUInt))
    TAMQPByteArray(len, read!(io, Vector{UInt8}(undef, len)))
end

write(io::IO, s::T) where {T<:Union{TAMQPShortStr,TAMQPLongStr,TAMQPByteArray}} = write(io, hton(s.len), s.data)

function read(io::IO, ::Type{TAMQPFieldValue})
    c = read(io, Char)
    v = read(io, FieldValueIndicatorMap[c])
    T = FieldValueIndicatorMap[c]
    TAMQPFieldValue{T}(c, v)
end

write(io::IO, fv::TAMQPFieldValue) = write(io, fv.typ, fv.fld)

read(io::IO, ::Type{TAMQPFieldValuePair}) = TAMQPFieldValuePair(read(io, TAMQPFieldName), read(io, TAMQPFieldValue))

write(io::IO, fv::TAMQPFieldValuePair) = write(io, fv.name, fv.val)

function read(io::IO, ::Type{TAMQPFieldTable})
    len = ntoh(read(io, fieldtype(TAMQPFieldTable, :len)))
    @debug("read fieldtable", len)
    buff = read!(io, Vector{UInt8}(undef, len))
    data = TAMQPFieldValuePair[]
    iob = IOBuffer(buff)
    while !eof(iob)
        push!(data, read(iob, TAMQPFieldValuePair))
    end
    TAMQPFieldTable(len, data)
end

function write(io::IO, ft::TAMQPFieldTable)
    @debug("write fieldtable", nfields=length(ft.data))
    iob = IOBuffer()
    for fv in ft.data
        write(iob, fv)
    end
    buff = take!(iob)
    len = TAMQPLongUInt(length(buff))
    @debug("write fieldtable", len)
    l = write(io, hton(len))
    if len > 0
        l += write(io, buff)
    end
    l
end

"""
Read a generic frame. All frames have the following wire format:

0      1         3      7                  size+7     size+8
+------+---------+---------+ +-------------+ +-----------+
| type | channel | size    | |    payload  | | frame-end |
+------+---------+---------+ +-------------+ +-----------+
octet    short     long       'size' octets      octet
"""
function read(io::IO, ::Type{TAMQPGenericFrame})
    hdr = ntoh(read(io, fieldtype(TAMQPGenericFrame, :hdr)))
    @assert hdr in (1,2,3,8)
    props = read(io, fieldtype(TAMQPGenericFrame, :props))
    @debug("reading generic frame", type=hdr, channel=props.channel, payloadsize=props.payloadsize)
    payload = read!(io, TAMQPBodyPayload(Vector{TAMQPOctet}(undef, props.payloadsize)))
    fend = ntoh(read(io, fieldtype(TAMQPGenericFrame, :fend)))
    @assert fend == FrameEnd
    TAMQPGenericFrame(hdr, props, payload, fend)
end

write(io::IO, f::TAMQPGenericFrame) = write(io, hton(f.hdr), f.props, f.payload, f.fend)

# """
# Given a generic frame, convert it to appropriate exact frame type.
# """
#function narrow_frame(f::TAMQPGenericFrame)
#    if f.hdr == FrameMethod
#        return TAMQPMethodFrame(f)
#    end
#    throw(AMQPProtocolException("Unknown frame type $(f.hdr)"))
#end

function method_name(payload::TAMQPMethodPayload)
    c = CLASS_MAP[payload.class]
    m = c.method_map[payload.method]
    #(c.name, m.name)
    string(c.name) * "." * string(m.name)
end

"""
Validate if the method frame is for the given class and method.
"""
function is_method(m::TAMQPMethodFrame, class::Symbol, method::Symbol)
    c = CLASS_MAP[m.payload.class]
    if c.name === class
        m = c.method_map[m.payload.method]
        return m.name === method
    end
    false
end

function method_key(classname::Symbol, methodname::Symbol)
    class = CLASSNAME_MAP[classname]
    method = CLASSMETHODNAME_MAP[classname,methodname]
    (FrameMethod, class.id, method.id)
end
frame_key(frame_type) = (UInt8(frame_type),)

# ----------------------------------------
# IO for types end
# ----------------------------------------

# ----------------------------------------
# Connection and Channel begin
# ----------------------------------------

const UNUSED_CHANNEL = -1
const DEFAULT_CHANNEL = 0
const DEFAULT_CHANNELMAX = 256
const DEFAULT_AUTH_PARAMS = Dict{String,Any}("MECHANISM"=>"AMQPLAIN", "LOGIN"=>"guest", "PASSWORD"=>"guest")

const CONN_STATE_CLOSED = 0
const CONN_STATE_OPENING = 1
const CONN_STATE_OPEN = 2
const CONN_STATE_CLOSING = 3
const CONN_MAX_QUEUED = 1024 #typemax(Int)
const DEFAULT_KEEPALIVE_SECS = 60

abstract type AbstractChannel end

function keepalive!(sock, enable::Bool; interval::Integer=DEFAULT_KEEPALIVE_SECS)
    @debug("setting tcp keepalive on tcp socket", enable, interval)
    err = ccall(:uv_tcp_keepalive, Cint, (Ptr{Nothing}, Cint, Cuint), sock.handle, enable, interval)
    if err != 0
        throw(AMQPProtocolException("error setting keepalive on socket to $enable with interval $interval"))
    end
    return sock
end

mutable struct Connection
    virtualhost::String
    host::String
    port::Int
    sock::Union{TCPSocket, BufferedTLSSocket, Nothing}

    properties::Dict{Symbol,Any}
    capabilities::Dict{String,Any}
    channelmax::TAMQPShortInt
    framemax::TAMQPLongInt
    heartbeat::TAMQPShortInt
    enable_heartbeat::Bool
    keepalive::Integer
    enable_keepalive::Bool

    state::UInt8
    sendq::Channel{TAMQPGenericFrame}
    sendlck::Channel{UInt8}
    channels::Dict{TAMQPChannel, AbstractChannel}

    sender::Union{Task, Nothing}
    receiver::Union{Task, Nothing}
    heartbeater::Union{Task, Nothing}

    heartbeat_time_server::Float64
    heartbeat_time_client::Float64

    function Connection(;
            virtualhost::String="/",
            host::String="localhost",
            port::Int=AMQP_DEFAULT_PORT,
            send_queue_size::Int=CONN_MAX_QUEUED,
            heartbeat::Integer=0,
            enable_heartbeat::Bool=true,
            keepalive::Integer=DEFAULT_KEEPALIVE_SECS,
            enable_keepalive::Bool=true,
        )
        sendq = Channel{TAMQPGenericFrame}(send_queue_size)
        sendlck = Channel{UInt8}(1)
        put!(sendlck, 1)
        new(virtualhost, host, port, nothing,
            Dict{Symbol,Any}(), Dict{String,Any}(), 0, 0,
            heartbeat, enable_heartbeat, keepalive, enable_keepalive,
            CONN_STATE_CLOSED, sendq, sendlck, Dict{TAMQPChannel, AbstractChannel}(),
            nothing, nothing, nothing,
            0.0, 0.0)
    end
end

mutable struct MessageConsumer
    chan_id::TAMQPChannel
    consumer_tag::String
    recvq::Channel{Message}
    callback::Function
    receiver::Task

    function MessageConsumer(chan_id::TAMQPChannel, consumer_tag::String, callback::Function;
        buffer_size::Int=typemax(Int),
        buffer::Channel{Message}=Channel{Message}(buffer_size))

        c = new(chan_id, consumer_tag, buffer, callback)
        c.receiver = @async connection_processor(c, "Consumer $consumer_tag", channel_message_consumer)
        c
    end
end

close(consumer::MessageConsumer) = close(consumer.recvq)

mutable struct MessageChannel <: AbstractChannel
    id::TAMQPChannel
    conn::Connection
    state::UInt8
    flow::Bool

    recvq::Channel{TAMQPGenericFrame}
    receiver::Union{Task, Nothing}
    callbacks::Dict{Tuple,Tuple{Function,Any}}

    partial_msgs::Vector{Message} # holds partial messages while they are getting read (message bodies arrive in sequence)
    chan_get::Channel{Union{Message, Nothing}}  # channel used for received messages, in sync get call (TODO: maybe type more strongly?)
    consumers::Dict{String,MessageConsumer}
    pending_msgs::Dict{String,Channel{Message}} # holds messages received that do not have a consumer registered
    lck::ReentrantLock

    closereason::Union{CloseReason, Nothing}

    function MessageChannel(id, conn)
        new(id, conn, CONN_STATE_CLOSED, true,
            Channel{TAMQPGenericFrame}(CONN_MAX_QUEUED), nothing, Dict{Tuple,Tuple{Function,Any}}(),
            Message[], Channel{Union{Message, Nothing}}(1), Dict{String,MessageConsumer}(),
            Dict{String,Channel{Message}}(), ReentrantLock(), nothing)
    end
end

flush(c::MessageChannel) = flush(c.conn)
function flush(c::Connection)
    while isready(c.sendq) && (c.sender !== nothing) && !istaskdone(c.sender)
        yield()
    end
end

sock(c::MessageChannel) = sock(c.conn)
sock(c::Connection) = c.sock

isopen(c::Connection) = c.sock !== nothing && isopen(c.sock)
isopen(c::MessageChannel) = isopen(c.conn) && (c.id in keys(c.conn.channels))

get_property(c::MessageChannel, s::Symbol, default) = get_property(c.conn, s, default)
get_property(c::Connection, s::Symbol, default) = get(c.properties, s, default)

with_sendlock(f, c::MessageChannel) = with_sendlock(f, c.conn)
with_sendlock(f, c::Connection) = with_sendlock(f, c.sendlck)
function with_sendlock(f, sendlck::Channel{UInt8})
    lck = take!(sendlck)
    try
        f()
    finally
        put!(sendlck, lck)
    end
end
send(c::MessageChannel, f) = send(c.conn, f)
send(c::Connection, f) = put!(c.sendq, TAMQPGenericFrame(f))
function send(c::MessageChannel, payload::TAMQPMethodPayload)
    @debug("sending without content", methodname=method_name(payload))
    frameprop = TAMQPFrameProperties(c.id,0)
    send(c, TAMQPMethodFrame(frameprop, payload))
end
function send(c::MessageChannel, payload::TAMQPMethodPayload, msg::Message)
    @debug("sending with content", methodname=method_name(payload))
    frameprop = TAMQPFrameProperties(c.id,0)
    framemax = c.conn.framemax
    if framemax <= 0
        errormsg = (c.conn.state == CONN_STATE_OPEN) ? "Unexpected framemax ($framemax) value for connection" : "Connection closed"
        throw(AMQPClientException(errormsg))
    end

    with_sendlock(c) do
        send(c, TAMQPMethodFrame(frameprop, payload))
        hdrpayload = TAMQPHeaderPayload(payload.class, msg)
        send(c, TAMQPContentHeaderFrame(frameprop, hdrpayload))

        # send one or more message body frames
        offset = 1
        msglen = length(msg.data)
        @debug("sending message with content body", msglen)
        while offset <= msglen
            msgend = min(msglen, offset + framemax - 1)
            bodypayload = TAMQPBodyPayload(msg.data[offset:msgend])
            offset = msgend + 1
            @debug("sending content body frame", msglen, offset)
            send(c, TAMQPContentBodyFrame(frameprop, bodypayload))
        end
    end
end

# ----------------------------------------
# Async message handler framework begin
# ----------------------------------------
function wait_for_state(c, states; interval=1, timeout=typemax(Int))
    timedwait(Float64(timeout); pollint=Float64(interval)) do
        # if we are looking for open states, and connection gets closed in the meantime, it's an error, break out
        conn_error = !(CONN_STATE_CLOSED in states) && (c.state == CONN_STATE_CLOSED)
        state_found = (c.state in states)
        conn_error || state_found
    end
    c.state in states
end

function connection_processor(c, name, fn)
    @debug("Starting task", name)
    try
        while true
            fn(c)
        end
    catch err
        reason = "$name task exiting."
        if isa(c, MessageConsumer)
            if !(c.state in (CONN_STATE_CLOSING, CONN_STATE_CLOSED))
                reason = reason * " Unhandled exception: $err"
                @warn(reason, exception=(err,catch_backtrace()))
            end
            close(c)
        else
            isconnclosed = !isopen(c)
            ischanclosed = isa(c, MessageChannel) && isa(err, InvalidStateException) && err.state == :closed
            if ischanclosed || isconnclosed
                reason = reason * " Connection closed"
                if c.state !== CONN_STATE_CLOSING
                    reason = reason * " by peer"
                    close(c, false, true)
                end
                @debug(reason, exception=(err,catch_backtrace()))
            else
                if !(c.state in (CONN_STATE_CLOSING, CONN_STATE_CLOSED))
                    reason = reason * " Unhandled exception: $err"
                    @warn(reason, exception=(err,catch_backtrace()))
                end
                close(c, false, true)
            end
        end
    end
end

function connection_sender(c::Connection)
    @debug("==> sending on conn", host=c.host, port=c.port, virtualhost=c.virtualhost)
    nbytes = sendq_to_stream(sock(c), c.sendq)
    @debug("==> sent", nbytes)
    c.heartbeat_time_client = time()  # update heartbeat time for client
    nothing
end

function sendq_to_stream(conn::TCPSocket, sendq::Channel{TAMQPGenericFrame})
    msg = take!(sendq)
    if length(msg.payload.data) > TCP_MIN_WRITEBUFF_SIZE    # write large messages directly
        nbytes = write(conn, msg)
    else    # coalesce short messages and do single write
        buff = IOBuffer()
        nbytes = write(buff, msg)
        while isready(sendq) && (nbytes < TCP_MAX_WRITEBUFF_SIZE)
            nbytes += write(buff, take!(sendq))
        end
        write(conn, take!(buff))
    end
    nbytes
end
function sendq_to_stream(conn::BufferedTLSSocket, sendq::Channel{TAMQPGenericFrame})
    # avoid multiple small writes to TLS layer
    nbytes = write(conn, take!(sendq))
    while isready(sendq) && (nbytes < MbedTLS.MBEDTLS_SSL_MAX_CONTENT_LEN)
        nbytes += write(conn, take!(sendq))
    end
    # flush does a single write of accumulated buffer
    flush(conn)
    nbytes
end

function connection_receiver(c::Connection)
    f = read(sock(c), TAMQPGenericFrame)

    # update heartbeat time for server
    c.heartbeat_time_server = time()

    channelid = f.props.channel
    @debug("<== read message on conn", host=c.virtualhost, channelid)
    if !(channelid in keys(c.channels))
        @warn("Discarding message for unknown channel", channelid)
    end
    chan = channel(c, channelid)
    put!(chan.recvq, f)
    nothing
end

function connection_heartbeater(c::Connection)
    sleep(c.heartbeat)

    isopen(c) || throw(AMQPClientException("Connection closed"))

    now = time()
    if (now - c.heartbeat_time_client) > c.heartbeat
        send_connection_heartbeat(c)
    end

    if (now - c.heartbeat_time_server) > (2 * c.heartbeat)
        @warn("server heartbeat missed", secs=(now - c.heartbeat_time_server))
        close(c, false, false)
    end
    nothing
end

function channel_receiver(c::MessageChannel)
    f = take!(c.recvq)
    if f.hdr == FrameMethod
        m = TAMQPMethodFrame(f)
        @debug("<== received", channel=f.props.channel, class=m.payload.class, method=m.payload.method)
        cbkey = (f.hdr, m.payload.class, m.payload.method)
    elseif f.hdr == FrameHeartbeat
        m = TAMQPHeartBeatFrame(f)
        @debug("<== received heartbeat", channel=f.props.channel)
        cbkey = (f.hdr,)
    elseif f.hdr == FrameHeader
        m = TAMQPContentHeaderFrame(f)
        @debug("<== received contentheader", channel=f.props.channel)
        cbkey = (f.hdr,)
    elseif f.hdr == FrameBody
        m = TAMQPContentBodyFrame(f)
        @debug("<== received contentbody", channel=f.props.channel)
        cbkey = (f.hdr,)
    else
        m = f
        @warn("<== received unhandled frame type", channel=f.props.channel, type=f.hdr)
        cbkey = (f.hdr,)
    end
    (cb,ctx) = get(c.callbacks, cbkey, (on_unexpected_message, nothing))
    @assert f.props.channel == c.id
    cb(c, m, ctx)
    nothing
end

function channel_message_consumer(c::MessageConsumer)
    m = take!(c.recvq)
    c.callback(m)
    nothing
end

clear_handlers(c::MessageChannel) = (empty!(c.callbacks); nothing)
function handle(c::MessageChannel, classname::Symbol, methodname::Symbol, cb=nothing, ctx=nothing)
    cbkey = method_key(classname, methodname)
    if cb === nothing
        delete!(c.callbacks, cbkey)
    else
        c.callbacks[cbkey] = (cb, ctx)
    end
    nothing
end
function handle(c::MessageChannel, frame_type::Integer, cb=nothing, ctx=nothing)
    cbkey = frame_key(frame_type)
    if cb === nothing
        delete!(c.callbacks, cbkey)
    else
        c.callbacks[cbkey] = (cb, ctx)
    end
    nothing
end

# ----------------------------------------
# Async message handler framework end
# ----------------------------------------

# ----------------------------------------
# Open channel / connection begin
# ----------------------------------------

function find_unused_channel(c::Connection)
    k = keys(c.channels)
    maxid = c.channelmax
    for id in 0:maxid
        if !(id in k)
            return id
        end
    end
    throw(AMQPClientException("No free channel available (max: $maxid)"))
end

"""
    channel(conn, id, create)
    channel(f, args...)

Create or return an existing a channel object.
Multiple channels can be multiplexed over a single connection.
Can be used with the Julia do block syntax to create a channel and close it afterwards.

- `conn`: The connection over which to create the channel.
- `id`: Channels are identified by their numeric id. Specifying `AMQPClient.UNUSED_CHANNEL` as channel
    id during creation will automatically assign an unused id.
- `create`: If true, a new channel will be created. Else an existing channel with the specified id
    will be returned.
"""
channel(c::MessageChannel, id::Integer) = channel(c.conn, id)
channel(c::Connection, id::Integer) = c.channels[id]
channel(c::MessageChannel, id::Integer, create::Bool) = channel(c.conn, id, create)
function channel(c::Connection, id::Integer, create::Bool; connect_timeout=DEFAULT_CONNECT_TIMEOUT)
    if create
        if id == UNUSED_CHANNEL
            id = find_unused_channel(c)
        elseif id in keys(c.channels)
            throw(AMQPClientException("Channel Id $id is already in use"))
        end
        chan = MessageChannel(id, c)
        chan.state = CONN_STATE_OPENING
        c.channels[chan.id] = chan

        if id != DEFAULT_CHANNEL
            # open the channel
            chan.receiver = @async connection_processor(chan, "ChannelReceiver($(chan.id))", channel_receiver)
            handle(chan, :Channel, :OpenOk, on_channel_open_ok)
            send_channel_open(chan)

            if !wait_for_state(chan, CONN_STATE_OPEN; timeout=connect_timeout)
                error_message = "Channel handshake failed"
                if nothing !== chan.closereason
                    error_message = string(error_message, " - ", string(chan.closereason.code), " (", convert(String, chan.closereason.msg), ")")
                end
                throw(AMQPClientException(error_message))
            end
        end
    else
        chan = channel(c, id)
    end
    chan
end
function channel(f, args...; kwargs...)
    chan = channel(args...; kwargs...)
    try
        f(chan)
    catch
        rethrow()
    finally
        close(chan)
    end
end

"""
    connection(f; kwargs...)

    connection(;
        virtualhost = "/",
        host = "localhost",
        port = AMQPClient.AMQP_DEFAULT_PORT,
        framemax = 0,
        heartbeat = true,
        keepalive = DEFAULT_KEEPALIVE_SECS,
        send_queue_size = CONN_MAX_QUEUED,
        auth_params = AMQPClient.DEFAULT_AUTH_PARAMS,
        channelmax = AMQPClient.DEFAULT_CHANNELMAX,
        connect_timeout = AMQPClient.DEFAULT_CONNECT_TIMEOUT,
        amqps = nothing
    )

Creates a fresh connection to the AMQP server.
Returns a connection that can be used to open channels subsequently.
Can be used with the Julia do block syntax to create a connection and close it afterwards.

Keyword arguments:
- `host`: The message server host to connect to. Defaults to "localhost".
- `port`: The message server port to connect to. Defaults to the default AMQP port.
- `virtualhost`: The virtual host to connect to. Defaults to "/".
- `amqps`: If connection is to be done over AMQPS, the TLS options to use. See `amqps_configure`.
- `connect_timeout`: TCP connect timeout to impose. Default `AMQPClient.DEFAULT_CONNECT_TIMEOUT`,
- `framemax`: The maximum frame size to use. Defaults to 0, which means no limit.
- `heartbeat`: `true` to enable heartbeat, `false` to disable. Can also be set to a positive integer,
    in which case it is the heartbeat interval in seconds. Defaults to `true`. If `false`, ensure
    `keepalive` is enabled to detect dead connections. This parameter is negotiated with the server.
- `keepalive`: `true` to enable TCP keepalives, `false` to disable. Can also be set to a positive integer,
    in which case it is the keepalive interval in seconds. Defaults to `DEFAULT_KEEPALIVE_SECS`.
- `send_queue_size`: Maximum number of items to buffer in memory before blocking the send API until
    messages are drained. Defaults to CONN_MAX_QUEUED.
- `auth_params`: Parameters to use to authenticate the connection. Defaults to AMQPClient.DEFAULT_AUTH_PARAMS.
- `channelmax`: Maximum channel number to impose/negotiate with the server. Defaults to AMQPClient.DEFAULT_CHANNELMAX.

"""
function connection(; virtualhost="/", host="localhost", port=AMQPClient.AMQP_DEFAULT_PORT,
        framemax=0,
        heartbeat::Union{Int,Bool}=true,
        keepalive::Union{Int,Bool}=DEFAULT_KEEPALIVE_SECS,
        send_queue_size::Integer=CONN_MAX_QUEUED,
        auth_params=AMQPClient.DEFAULT_AUTH_PARAMS,
        channelmax::Integer=AMQPClient.DEFAULT_CHANNELMAX,
        connect_timeout=AMQPClient.DEFAULT_CONNECT_TIMEOUT,
        amqps::Union{MbedTLS.SSLConfig,Nothing}=nothing)
    @debug("connecting", host, port, virtualhost)

    keepalive_interval = isa(keepalive, Bool) ? DEFAULT_KEEPALIVE_SECS : keepalive
    enable_keepalive = isa(keepalive, Bool) ? keepalive : (keepalive_interval > 0)

    heartbeat_interval = isa(heartbeat, Bool) ? 0 : heartbeat
    enable_heartbeat = isa(heartbeat, Bool) ? heartbeat : (heartbeat > 0)

    conn = Connection(;
                virtualhost=virtualhost,
                host=host,
                port=port,
                send_queue_size=send_queue_size,
                heartbeat=heartbeat_interval,
                enable_heartbeat=enable_heartbeat,
                keepalive=keepalive_interval,
                enable_keepalive=enable_keepalive,)
    chan = channel(conn, AMQPClient.DEFAULT_CHANNEL, true)

    # setup handler for Connection.Start
    ctx = Dict(:auth_params=>auth_params, :channelmax=>channelmax, :framemax=>framemax, :heartbeat=>heartbeat_interval)
    AMQPClient.handle(chan, :Connection, :Start, AMQPClient.on_connection_start, ctx)

    # open socket and start processor tasks
    sock = connect(conn.host, conn.port)
    isdefined(Sockets, :nagle) && Sockets.nagle(sock, false)
    isdefined(Sockets, :quickack) && Sockets.quickack(sock, true)
    keepalive!(sock, enable_keepalive; interval=keepalive_interval)

    conn.sock = (amqps !== nothing) ? setup_tls(sock, host, amqps) : sock
    conn.sender = @async   AMQPClient.connection_processor(conn, "ConnectionSender", AMQPClient.connection_sender)
    conn.receiver = @async AMQPClient.connection_processor(conn, "ConnectionReceiver", AMQPClient.connection_receiver)
    chan.receiver = @async AMQPClient.connection_processor(chan, "ChannelReceiver($(chan.id))", AMQPClient.channel_receiver)

    # initiate handshake
    conn.state = chan.state = AMQPClient.CONN_STATE_OPENING
    write(AMQPClient.sock(chan), AMQPClient.ProtocolHeader)
    flush(AMQPClient.sock(chan))

    if !AMQPClient.wait_for_state(conn, AMQPClient.CONN_STATE_OPEN; timeout=connect_timeout) || !AMQPClient.wait_for_state(chan, AMQPClient.CONN_STATE_OPEN; timeout=connect_timeout)
        error_message = "Connection handshake failed"
        if nothing !== chan.closereason
            error_message = string(error_message, " - ", string(chan.closereason.code), " (", convert(String, chan.closereason.msg), ")")
        end
        throw(AMQPClientException(error_message))
    end
    chan
end

function connection(f; kwargs...)
    conn = connection(; kwargs...)

    try
        f(conn)
    catch
        rethrow()
    finally
        close(conn)
    end
end

# ----------------------------------------
# Open channel / connection end
# ----------------------------------------

# ----------------------------------------
# Close channel / connection begin
# ----------------------------------------

function close(chan::MessageChannel, handshake::Bool=true, by_peer::Bool=false, reply_code=ReplySuccess, reply_text="", class_id=0, method_id=0)
    (chan.state == CONN_STATE_CLOSED) && (return nothing)

    conn = chan.conn

    if chan.id == DEFAULT_CHANNEL
        # default channel represents the connection
        close(conn, handshake, by_peer, reply_code, reply_text, class_id, method_id)
    elseif chan.state != CONN_STATE_CLOSING
        # send handshake if needed and when called the first time
        chan.state = CONN_STATE_CLOSING
        if handshake && !by_peer
            send_channel_close(chan, reply_code, reply_text, class_id, method_id)
        end
    end

    # release resources when closed by peer or when closing abruptly
    if !handshake || by_peer
        close(chan.recvq)
        close(chan.chan_get)
        map(close, values(chan.consumers))
        empty!(chan.consumers)
        chan.receiver = nothing
        chan.callbacks = Dict{Tuple,Tuple{Function,Any}}()
        delete!(chan.conn.channels, chan.id)
        chan.state = CONN_STATE_CLOSED
    end
    nothing
end

function close(conn::Connection, handshake::Bool=true, by_peer::Bool=false, reply_code=ReplySuccess, reply_text="", class_id=0, method_id=0)
    (conn.state == CONN_STATE_CLOSED) && (return nothing)

    # send handshake if needed and when called the first time
    if conn.state != CONN_STATE_CLOSING
        conn.state = CONN_STATE_CLOSING

        # close all other open channels
        for open_channel in collect(values(conn.channels))
            if open_channel.id != DEFAULT_CHANNEL
                close(open_channel, false, by_peer)
            end
        end

        # send handshake if needed
        if handshake && !by_peer
            send_connection_close(conn, reply_code, reply_text, class_id, method_id)
        end
    end

    if !handshake || by_peer
        # close socket
        close(conn.sock)
        conn.sock = nothing

        # reset all members
        conn.properties = Dict{Symbol,Any}()
        conn.capabilities = Dict{String,Any}()
        conn.channelmax = 0
        conn.framemax = 0
        conn.heartbeat = 0

        # close and reset the sendq channel
        close(conn.sendq)
        conn.sendq = Channel{TAMQPGenericFrame}(CONN_MAX_QUEUED)

        # reset the tasks
        conn.sender = nothing
        conn.receiver = nothing
        conn.heartbeater = nothing
        conn.state = CONN_STATE_CLOSED
    end
    nothing
end

# ----------------------------------------
# Close channel / connection end
# ----------------------------------------

# ----------------------------------------
# Connection and Channel end
# ----------------------------------------

# ----------------------------------------
# Exchange begin
# ----------------------------------------
const EXCHANGE_TYPE_DIRECT = "direct"      # must be implemented by servers
const EXCHANGE_TYPE_FANOUT = "fanout"      # must be implemented by servers
const EXCHANGE_TYPE_TOPIC = "topic"        # optional, must test before typing to open
const EXCHANGE_TYPE_HEADERS = "headers"    # optional, must test before typing to open

# The server MUST, in each virtual host, predeclare an exchange instance for each standard
# exchange type that it implements, where the name of the exchange instance, if defined, is "amq."
# followed by the exchange type name.
# The server MUST predeclare a direct exchange with no public name to act as the default
# exchange for content Publish methods and for default queue bindings.
default_exchange_name(excg_type) = ("amq." * excg_type)
default_exchange_name() = ""

function _wait_resp(sendmethod, chan::MessageChannel, default_result::T,
        nowait::Bool=true, resp_handler=nothing, resp_class=nothing, resp_meth=nothing,
        timeout_result::T=default_result, timeout::Int=DEFAULT_TIMEOUT) where {T}
    result = default_result
    if !nowait
        reply = Channel{T}(1)
        # register a callback
        handle(chan, resp_class, resp_meth, resp_handler, reply)
    end

    sendmethod()

    if !nowait
        # wait for response
        result = timeout_result
        if :ok === timedwait(()->(isready(reply) || !isopen(chan)), Float64(timeout); pollint=0.01)
            if isready(reply)
                result = take!(reply)
            else
                error_message = "Connection closed"
                if nothing !== chan.closereason
                    error_message = string(error_message, " - ", string(chan.closereason.code), " (", convert(String, chan.closereason.msg), ")")
                end
                throw(AMQPClientException(error_message))
            end
        end
        close(reply)
    end
    result
end

function exchange_declare(chan::MessageChannel, name::String, typ::String;
        passive::Bool=false, durable::Bool=false, auto_delete::Bool=false,
        nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT,
        arguments::Dict{String,Any}=Dict{String,Any}())
    (isempty(name) || startswith(name, "amq.")) && !passive && throw(AMQPClientException("Exchange name '$name' is reserved. Use a different name."))
    if auto_delete
        @debug("Warning: auto_delete exchange types are deprecated")
    end

    _wait_resp(chan, true, nowait, on_exchange_declare_ok, :Exchange, :DeclareOk, false, timeout) do
        send_exchange_declare(chan, name, typ, passive, durable, auto_delete, nowait, arguments)
    end
end

function exchange_delete(chan::MessageChannel, name::String; if_unused::Bool=false, nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT)
    (isempty(name) || startswith(name, "amq.")) && throw(AMQPClientException("Exchange name '$name' is reserved. Use a different name."))
    _wait_resp(chan, true, nowait, on_exchange_delete_ok, :Exchange, :DeleteOk, false, timeout) do
        send_exchange_delete(chan, name, if_unused, nowait)
    end
end

function exchange_bind(chan::MessageChannel, dest::String, src::String, routing_key::String;
        nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT,
        arguments::Dict{String,Any}=Dict{String,Any}())
    _wait_resp(chan, true, nowait, on_exchange_bind_ok, :Exchange, :BindOk, false, timeout) do
        send_exchange_bind(chan, dest, src, routing_key, nowait, arguments)
    end
end

function exchange_unbind(chan::MessageChannel, dest::String, src::String, routing_key::String;
        nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT,
        arguments::Dict{String,Any}=Dict{String,Any}())
    _wait_resp(chan, true, nowait, on_exchange_unbind_ok, :Exchange, :UnbindOk, false, timeout) do
        send_exchange_unbind(chan, dest, src, routing_key, nowait, arguments)
    end
end

# ----------------------------------------
# Exchange end
# ----------------------------------------

# ----------------------------------------
# Queue begin
# ----------------------------------------
"""Declare a queue (or query an existing queue).
Returns a tuple: (boolean success/failure, queue name, message count, consumer count)
"""
function queue_declare(chan::MessageChannel, name::String;
        passive::Bool=false, durable::Bool=false, exclusive::Bool=false, auto_delete::Bool=false,
        nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT,
        arguments::Dict{String,Any}=Dict{String,Any}())
    _wait_resp(chan, (true, "", TAMQPMessageCount(0), Int32(0)), nowait, on_queue_declare_ok, :Queue, :DeclareOk, (false,"", TAMQPMessageCount(0), Int32(0)), timeout) do
        send_queue_declare(chan, name, passive, durable, exclusive, auto_delete, nowait, arguments)
    end
end

function queue_bind(chan::MessageChannel, queue_name::String, excg_name::String, routing_key::String; nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT, arguments::Dict{String,Any}=Dict{String,Any}())
    _wait_resp(chan, true, nowait, on_queue_bind_ok, :Queue, :BindOk, false, timeout) do
        send_queue_bind(chan, queue_name, excg_name, routing_key, nowait, arguments)
    end
end

function queue_unbind(chan::MessageChannel, queue_name::String, excg_name::String, routing_key::String; arguments::Dict{String,Any}=Dict{String,Any}(), timeout::Int=DEFAULT_TIMEOUT)
    nowait = false
    _wait_resp(chan, true, nowait, on_queue_unbind_ok, :Queue, :UnbindOk, false, timeout) do
        send_queue_unbind(chan, queue_name, excg_name, routing_key, arguments)
    end
end

"""Purge messages from a queue.
Returns a tuple: (boolean success/failure, message count)
"""
function queue_purge(chan::MessageChannel, name::String; nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT)
    _wait_resp(chan, (true,TAMQPMessageCount(0)), nowait, on_queue_purge_ok, :Queue, :PurgeOk, (false,TAMQPMessageCount(0)), timeout) do
        send_queue_purge(chan, name, nowait)
    end
end

"""Delete a queue.
Returns a tuple: (boolean success/failure, message count)
"""
function queue_delete(chan::MessageChannel, name::String; if_unused::Bool=false, if_empty::Bool=false, nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT)
    _wait_resp(chan, (true,TAMQPMessageCount(0)), nowait, on_queue_delete_ok, :Queue, :DeleteOk, (false,TAMQPMessageCount(0)), timeout) do
        send_queue_delete(chan, name, if_unused, if_empty, nowait)
    end
end

# ----------------------------------------
# Queue end
# ----------------------------------------

# ----------------------------------------
# Tx begin
# ----------------------------------------

function _tx(sendmethod, chan::MessageChannel, respmethod::Symbol, on_resp, timeout::Int)
    nowait = false
    _wait_resp(chan, true, nowait, on_resp, :Tx, respmethod, false, timeout) do
        sendmethod(chan)
    end
end

tx_select(chan::MessageChannel; timeout::Int=DEFAULT_TIMEOUT) = _tx(send_tx_select, chan, :SelectOk, on_tx_select_ok, timeout)
tx_commit(chan::MessageChannel; timeout::Int=DEFAULT_TIMEOUT) = _tx(send_tx_commit, chan, :CommitOk, on_tx_commit_ok, timeout)
tx_rollback(chan::MessageChannel; timeout::Int=DEFAULT_TIMEOUT) = _tx(send_tx_rollback, chan, :RollbackOk, on_tx_rollback_ok, timeout)

# ----------------------------------------
# Tx end
# ----------------------------------------

# ----------------------------------------
# Basic begin
# ----------------------------------------

function basic_qos(chan::MessageChannel, prefetch_size, prefetch_count, apply_global::Bool; timeout::Int=DEFAULT_TIMEOUT)
    nowait = false
    _wait_resp(chan, true, nowait, on_basic_qos_ok, :Basic, :QosOk, false, timeout) do
        send_basic_qos(chan, prefetch_size, prefetch_count, apply_global)
    end
end

"""Start a queue consumer.

queue: queue name
consumer_tag: id of the consumer, server generates a unique tag if this is empty
no_local: do not deliver own messages
no_ack: no acknowledgment needed, server automatically and silently acknowledges delivery (speed at the cost of reliability)
exclusive: request exclusive access (only this consumer can access the queue)
nowait: do not send a reply method
"""
function basic_consume(chan::MessageChannel, queue::String, consumer_fn::Function; consumer_tag::String="", no_local::Bool=false, no_ack::Bool=false,
    exclusive::Bool=false, nowait::Bool=false, arguments::Dict{String,Any}=Dict{String,Any}(), timeout::Int=DEFAULT_TIMEOUT, buffer_sz::Int=typemax(Int))

    # register the consumer and get the consumer_tag
    result = _wait_resp(chan, (true, ""), nowait, on_basic_consume_ok, :Basic, :ConsumeOk, (false, ""), timeout) do
        send_basic_consume(chan, queue, consumer_tag, no_local, no_ack, exclusive, nowait, arguments)
    end

    # start the message consumer
    if result[1]
        consumer_tag = result[2]

        # set up message buffer beforehand to store messages that the consumer may receive while we are still setting things up,
        # or get the buffer that was set up already because we received messages
        lock(chan.lck) do
            consumer_buffer = get!(chan.pending_msgs, consumer_tag) do
                Channel{Message}(buffer_sz)
            end
            consumer_buffer.sz_max = buffer_sz
            chan.consumers[consumer_tag] = MessageConsumer(chan.id, consumer_tag, consumer_fn; buffer=consumer_buffer)
            delete!(chan.pending_msgs, consumer_tag)
        end
    end

    result
end

"""Cancels a consumer.

This does not affect already delivered messages, but it does mean the server will not send any more messages for that consumer. The client may receive an arbitrary number of
messages in between sending the cancel method and receiving the cancelok reply.
"""
function basic_cancel(chan::MessageChannel, consumer_tag::String; nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT)
    result = _wait_resp(chan, (true, ""), nowait, on_basic_cancel_ok, :Basic, :CancelOk, (false, ""), timeout) do
        send_basic_cancel(chan, consumer_tag, nowait)
    end

    # clear a message consumer
    if result[1]
        if consumer_tag in keys(chan.consumers)
            close(chan.consumers[consumer_tag])
            delete!(chan.consumers, consumer_tag)
        end
    end
    result[1]
end

"""Publish a message

This method publishes a message to a specific exchange. The message will be routed to queues as defined by the exchange
configuration and distributed to any active consumers when the transaction, if any, is committed.
"""
function basic_publish(chan::MessageChannel, msg::Message; exchange::String="", routing_key::String="", mandatory::Bool=false, immediate::Bool=false)
    send_basic_publish(chan, msg, exchange, routing_key, mandatory, immediate)
end

const GET_EMPTY_RESP = nothing
function basic_get(chan::MessageChannel, queue::String, no_ack::Bool)
    send_basic_get(chan, queue, no_ack)
    take!(chan.chan_get)
end

basic_ack(chan::MessageChannel, delivery_tag::TAMQPDeliveryTag; all_upto::Bool=false) = send_basic_ack(chan, delivery_tag, all_upto)
basic_reject(chan::MessageChannel, delivery_tag::TAMQPDeliveryTag; requeue::Bool=false) = send_basic_reject(chan, delivery_tag, requeue)

function basic_recover(chan::MessageChannel, requeue::Bool=false; async::Bool=false, timeout::Int=DEFAULT_TIMEOUT)
    _wait_resp(chan, true, async, on_basic_recover_ok, :Basic, :RecoverOk, false, timeout) do
        send_basic_recover(chan, requeue, async)
    end
end

# ----------------------------------------
# Basic end
# ----------------------------------------

# ----------------------------------------
# Confirm begin
# ----------------------------------------

function confirm_select(chan::MessageChannel; nowait::Bool=false, timeout::Int=DEFAULT_TIMEOUT)
    _wait_resp(chan, true, nowait, on_confirm_select_ok, :Confirm, :SelectOk, false, timeout) do
        send_confirm_select(chan)
    end
end

send_confirm_select(chan::MessageChannel) = send(chan, TAMQPMethodPayload(:Confirm, :Select, ()))

# ----------------------------------------
# Confirm end
# ----------------------------------------

# ----------------------------------------
# send and recv for methods begin
# ----------------------------------------

function on_unexpected_message(c::MessageChannel, m::TAMQPMethodFrame, ctx)
    @debug("Unexpected message", channel=c.id, class=m.payload.class, method=m.payload.method)
    nothing
end

function on_unexpected_message(c::MessageChannel, f, ctx)
    @debug("Unexpected message", channel=c.id, frametype=f.hdr)
    nothing
end

function _on_ack(chan::MessageChannel, m::TAMQPMethodFrame, class::Symbol, method::Symbol, ctx)
    @assert is_method(m, class, method)
    if ctx !== nothing
        put!(ctx, true)
    end
    handle(chan, class, method)
    nothing
end

_send_close_ok(context_class::Symbol, chan::MessageChannel) = send(chan, TAMQPMethodPayload(context_class, :CloseOk, ()))

function _on_close_ok(context_class::Symbol, chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, context_class, :CloseOk)
    close(chan, false, true)
    nothing
end

function _send_close(context_class::Symbol, chan::MessageChannel, reply_code=ReplySuccess, reply_text="", class_id=0, method_id=0)
    chan.closereason = CloseReason(TAMQPReplyCode(reply_code), TAMQPReplyText(reply_text), TAMQPClassId(class_id), TAMQPMethodId(method_id))
    if context_class === :Channel && chan.id == DEFAULT_CHANNEL
        @debug("closing channel 0 is equivalent to closing the connection!")
        context_class = :Connection
    end
    context_chan_id = context_class === :Connection ? 0 : chan.id

    _send_close(context_class, context_chan_id, chan.conn, reply_code, reply_text, class_id, method_id, chan.id)
end

_send_close(context_class::Symbol, context_chan_id, conn::Connection, reply_code=ReplySuccess, reply_text="", class_id=0, method_id=0, chan_id=0) =
    send(conn, TAMQPMethodFrame(TAMQPFrameProperties(context_chan_id,0), TAMQPMethodPayload(context_class, :Close, (TAMQPReplyCode(reply_code), TAMQPReplyText(reply_text), TAMQPClassId(class_id), TAMQPMethodId(method_id)))))

send_connection_close_ok(chan::MessageChannel) = _send_close_ok(:Connection, chan)
on_connection_close_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_close_ok(:Connection, chan, m, ctx)
function on_connection_close(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Connection, :Close)
    @assert chan.id == DEFAULT_CHANNEL
    chan.closereason = CloseReason(m.payload.fields[1].second, m.payload.fields[2].second, m.payload.fields[3].second, m.payload.fields[4].second)
    send_connection_close_ok(chan)
    t1 = time()
    while isready(chan.conn.sendq) && ((time() - t1) < 5)
        yield() # wait 5 seconds (arbirtary) for the message to get sent
    end
    close(chan, false, true)
end
function on_channel_close(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Channel, :Close)
    @assert chan.id != DEFAULT_CHANNEL
    chan.closereason = CloseReason(m.payload.fields[1].second, m.payload.fields[2].second, m.payload.fields[3].second, m.payload.fields[4].second)
    send_channel_close_ok(chan)
    close(chan, false, true)
end

send_connection_close(chan::MessageChannel, reply_code=ReplySuccess, reply_text="", class_id=0, method_id=0) = _send_close(:Connection, chan, reply_code, reply_text, class_id, method_id)
send_connection_close(conn::Connection, reply_code=ReplySuccess, reply_text="", class_id=0, method_id=0) = _send_close(:Connection, 0, conn, reply_code, reply_text, class_id, method_id)

send_channel_close_ok(chan::MessageChannel) = _send_close_ok(:Channel, chan)
on_channel_close_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_close_ok(:Channel, chan, m, ctx)
send_channel_close(chan::MessageChannel, reply_code=ReplySuccess, reply_text="", class_id=0, method_id=0) = _send_close(:Channel, chan, reply_code, reply_text, class_id, method_id)

function on_connection_start(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Connection, :Start)
    @assert chan.id == DEFAULT_CHANNEL
    conn = chan.conn

    # setup server properties and capabilities
    merge!(conn.properties, Dict{Symbol,Any}(Symbol(n)=>simplify(v) for (n,v) in m.payload.fields))
    server_props = simplify(get_property(chan, :ServerProperties, TAMQPFieldTable(Dict{String,Any}())))
    if "capabilities" in keys(server_props)
        merge!(conn.capabilities, server_props["capabilities"])
    end

    handle(chan, :Connection, :Start)
    auth_params = ctx[:auth_params]
    delete!(ctx, :auth_params)  # we cont need auth params any more
    handle(chan, :Connection, :Tune, on_connection_tune, ctx)
    send_connection_start_ok(chan, auth_params)
    nothing
end

function send_connection_start_ok(chan::MessageChannel, auth_params::Dict{String,Any})
    conn = chan.conn

    # set up client_props
    client_props = copy(CLIENT_IDENTIFICATION)
    client_cap = client_props["capabilities"]
    server_cap = conn.capabilities
    @debug("server capabilities", server_cap)
    if "consumer_cancel_notify" in keys(server_cap)
        client_cap["consumer_cancel_notify"] = server_cap["consumer_cancel_notify"]
    end
    if "connection.blocked" in keys(server_cap)
        client_cap["connection.blocked"] = server_cap["connection.blocked"]
    end
    @debug("client_props", client_props)

    # assert that auth mechanism is supported
    mechanism = auth_params["MECHANISM"]
    mechanisms = split(get_property(chan, :Mechanisms, ""), ' ')
    @debug("checking auth mechanism", mechanism, supported=mechanisms)
    @assert mechanism in mechanisms

    # set up locale
    # pick up one of the server locales
    locales = split(get_property(chan, :Locales, ""), ' ')
    @debug("supported locales", locales)
    client_locale = locales[1]
    @debug("client_locale", client_locale)

    # respond to login
    auth_resp = AUTH_PROVIDERS[mechanism](auth_params)
    @debug("auth_resp", auth_resp)

    send(chan, TAMQPMethodPayload(:Connection, :StartOk, (client_props, mechanism, auth_resp, client_locale)))
    nothing
end

function on_connection_tune(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Connection, :Tune)
    @assert chan.id == DEFAULT_CHANNEL
    conn = chan.conn

    conn.channelmax = m.payload.fields[1].second
    conn.framemax = m.payload.fields[2].second
    conn.heartbeat = m.payload.fields[3].second
    @debug("got_connection_tune", channelmax=conn.channelmax, framemax=conn.framemax, heartbeat=conn.heartbeat)
    handle(chan, FrameHeartbeat, on_connection_heartbeat)
    send_connection_tune_ok(chan, ctx[:channelmax], ctx[:framemax], ctx[:heartbeat])
    handle(chan, :Connection, :Tune)
    handle(chan, :Connection, :OpenOk, on_connection_open_ok, ctx)
    send_connection_open(chan)
    nothing
end

function send_connection_tune_ok(chan::MessageChannel, channelmax=0, framemax=0, heartbeat=0)
    conn = chan.conn

    # negotiate (min of what expected by both parties)
    function opt(desired_param, limited_param)
        if desired_param > 0 && limited_param > 0
            min(desired_param, limited_param)
        else
            max(desired_param, limited_param)
        end
    end

    conn.channelmax = opt(channelmax, conn.channelmax)
    conn.framemax = opt(framemax, conn.framemax)
    conn.heartbeat = conn.enable_heartbeat ? opt(heartbeat, conn.heartbeat) : 0

    @debug("send_connection_tune_ok", channelmax=conn.channelmax, framemax=conn.framemax, heartbeat=conn.heartbeat)
    send(chan, TAMQPMethodPayload(:Connection, :TuneOk, (conn.channelmax, conn.framemax, conn.heartbeat)))

    if conn.enable_heartbeat
        # start heartbeat timer
        conn.heartbeater = @async connection_processor(conn, "HeartBeater", connection_heartbeater)
    end
    nothing
end

send_connection_open(chan::MessageChannel) = send(chan, TAMQPMethodPayload(:Connection, :Open, (chan.conn.virtualhost, "", false)))

function on_connection_open_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Connection, :OpenOk)
    @assert chan.id == DEFAULT_CHANNEL
    conn = chan.conn

    conn.state = CONN_STATE_OPEN
    chan.state = CONN_STATE_OPEN

    handle(chan, :Connection, :Close, on_connection_close, ctx)
    handle(chan, :Connection, :CloseOk, on_connection_close_ok, ctx)
    handle(chan, :Connection, :OpenOk)

    nothing
end

send_connection_heartbeat(conn::Connection) = send(conn, TAMQPHeartBeatFrame())
on_connection_heartbeat(chan::MessageChannel, h::TAMQPHeartBeatFrame, ctx) = nothing

send_channel_open(chan::MessageChannel) = send(chan, TAMQPMethodPayload(:Channel, :Open, ("",)))
send_channel_flow(chan::MessageChannel, flow::Bool) = send(chan, TAMQPMethodPayload(:Channel, :Flow, (flow,)))

function on_channel_open_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    chan.state = CONN_STATE_OPEN
    handle(chan, :Channel, :Flow,       on_channel_flow, :Flow)
    handle(chan, :Channel, :FlowOk,     on_channel_flow, :FlowOk)
    handle(chan, :Channel, :Close,      on_channel_close)
    handle(chan, :Channel, :CloseOk,    on_channel_close_ok)
    handle(chan, :Basic,   :GetOk,      on_basic_get_empty_or_ok)
    handle(chan, :Basic,   :GetEmpty,   on_basic_get_empty_or_ok)
    handle(chan, :Basic,   :Deliver,    on_basic_get_empty_or_ok)
    handle(chan, FrameHeader, on_channel_message_in)
    handle(chan, FrameBody, on_channel_message_in)
    nothing
end

function on_channel_flow(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Channel, ctx)
    chan.flow = m.payload.fields[1].second
    @debug("on_channel_flow", channel=chan.id, flow=chan.flow)
    nothing
end

send_exchange_declare(chan::MessageChannel, name::String, typ::String, passive::Bool, durable::Bool, auto_delete::Bool, nowait::Bool, arguments::Dict{String,Any}) =
    send(chan, TAMQPMethodPayload(:Exchange, :Declare, (0, name, typ, passive, durable, auto_delete, false, nowait, arguments)))
send_exchange_delete(chan::MessageChannel, name::String, if_unused::Bool, nowait::Bool) = send(chan, TAMQPMethodPayload(:Exchange, :Delete, (0, name, if_unused, nowait)))
_send_exchange_bind_unbind(chan::MessageChannel, meth::Symbol, dest::String, src::String, routing_key::String, nowait::Bool, arguments::Dict{String,Any}) =
    send(chan, TAMQPMethodPayload(:Exchange, meth, (0, dest, src, routing_key, nowait, arguments)))
send_exchange_bind(chan::MessageChannel, dest::String, src::String, routing_key::String, nowait::Bool, arguments::Dict{String,Any}) = _send_exchange_bind_unbind(chan, :Bind, dest, src, routing_key, nowait, arguments)
send_exchange_unbind(chan::MessageChannel, dest::String, src::String, routing_key::String, nowait::Bool, arguments::Dict{String,Any}) = _send_exchange_bind_unbind(chan, :Unbind, dest, src, routing_key, nowait, arguments)

on_exchange_declare_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Exchange, :DeclareOk, ctx)
on_exchange_delete_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Exchange, :DeleteOk, ctx)
on_exchange_bind_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Exchange, :BindOk, ctx)
on_exchange_unbind_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Exchange, :UnbindOk, ctx)

send_queue_declare(chan::MessageChannel, name::String, passive::Bool, durable::Bool, exclusive::Bool, auto_delete::Bool, nowait::Bool, arguments::Dict{String,Any}) =
    send(chan, TAMQPMethodPayload(:Queue, :Declare, (0, name, passive, durable, exclusive, auto_delete, nowait, arguments)))
send_queue_bind(chan::MessageChannel, queue_name::String, excg_name::String, routing_key::String, nowait::Bool, arguments::Dict{String,Any}) =
    send(chan, TAMQPMethodPayload(:Queue, :Bind, (0, queue_name, excg_name, routing_key, nowait, arguments)))
send_queue_unbind(chan::MessageChannel, queue_name::String, excg_name::String, routing_key::String, arguments::Dict{String,Any}) = send(chan, TAMQPMethodPayload(:Queue, :Unbind, (0, queue_name, excg_name, routing_key, arguments)))
send_queue_purge(chan::MessageChannel, name::String, nowait::Bool) = send(chan, TAMQPMethodPayload(:Queue, :Purge, (0, name, nowait)))
send_queue_delete(chan::MessageChannel, name::String, if_unused::Bool, if_empty::Bool, nowait::Bool) = send(chan, TAMQPMethodPayload(:Queue, :Delete, (0, name, if_unused, if_empty, nowait)))

function on_queue_declare_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Queue, :DeclareOk)
    if ctx !== nothing
        name = convert(String, m.payload.fields[1].second)
        msg_count = m.payload.fields[2].second
        consumer_count = m.payload.fields[3].second
        put!(ctx, (true, name, msg_count, consumer_count))
    end
    handle(chan, :Queue, :DeclareOk)
    nothing
end

function _on_queue_purge_delete_ok(method::Symbol, chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Queue, method)
    if ctx !== nothing
        msg_count = m.payload.fields[1].second
        put!(ctx, (true, msg_count))
    end
    handle(chan, :Queue, method)
    nothing
end

on_queue_purge_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_queue_purge_delete_ok(:PurgeOk, chan, m, ctx)
on_queue_delete_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_queue_purge_delete_ok(:DeleteOk, chan, m, ctx)
on_queue_bind_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Queue, :BindOk, ctx)
on_queue_unbind_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Queue, :UnbindOk, ctx)

_send_tx(chan::MessageChannel, method::Symbol) = send(chan, TAMQPMethodPayload(:Tx, method, ()))
send_tx_select(chan::MessageChannel) = _send_tx(chan, :Select)
send_tx_commit(chan::MessageChannel) = _send_tx(chan, :Commit)
send_tx_rollback(chan::MessageChannel) = _send_tx(chan, :Rollback)

on_tx_select_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Tx, :SelectOk, ctx)
on_tx_commit_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Tx, :CommitOk, ctx)
on_tx_rollback_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Tx, :RollbackOk, ctx)

send_basic_qos(chan::MessageChannel, prefetch_size, prefetch_count, apply_global::Bool) = send(chan, TAMQPMethodPayload(:Basic, :Qos, (prefetch_size, prefetch_count, apply_global)))

send_basic_consume(chan::MessageChannel, queue::String, consumer_tag::String, no_local::Bool, no_ack::Bool, exclusive::Bool, nowait::Bool, arguments::Dict{String,Any}) =
    send(chan, TAMQPMethodPayload(:Basic, :Consume, (0, queue, consumer_tag, no_local, no_ack, exclusive, nowait, arguments)))

send_basic_cancel(chan::MessageChannel, consumer_tag::String, nowait::Bool) = send(chan, TAMQPMethodPayload(:Basic, :Cancel, (consumer_tag, nowait)))
send_basic_publish(chan::MessageChannel, msg::Message, exchange::String, routing_key::String, mandatory::Bool=false, immediate::Bool=false) =
    send(chan, TAMQPMethodPayload(:Basic, :Publish, (0, exchange, routing_key, mandatory, immediate)), msg)
send_basic_get(chan::MessageChannel, queue::String, no_ack::Bool) = send(chan, TAMQPMethodPayload(:Basic, :Get, (0, queue, no_ack)))
send_basic_ack(chan::MessageChannel, delivery_tag::TAMQPDeliveryTag, all_upto::Bool) = send(chan, TAMQPMethodPayload(:Basic, :Ack, (delivery_tag, all_upto)))
send_basic_reject(chan::MessageChannel, delivery_tag::TAMQPDeliveryTag, requeue::Bool) = send(chan, TAMQPMethodPayload(:Basic, :Reject, (delivery_tag, requeue)))
send_basic_recover(chan::MessageChannel, requeue::Bool, async::Bool) = send(chan, TAMQPMethodPayload(:Basic, async ? :RecoverAsync : :Recover, (requeue,)))

on_basic_qos_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Basic, :QosOk, ctx)
function _on_basic_consume_cancel_ok(method::Symbol, chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    @assert is_method(m, :Basic, method)
    if ctx !== nothing
        consumer_tag = convert(String, m.payload.fields[1].second)
        put!(ctx, (true, consumer_tag))
    end
    handle(chan, :Basic, method)
    nothing
end
on_basic_consume_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_basic_consume_cancel_ok(:ConsumeOk, chan, m, ctx)
on_basic_cancel_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_basic_consume_cancel_ok(:CancelOk, chan, m, ctx)
on_basic_recover_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Basic, :RecoverOk, ctx)

function on_basic_get_empty_or_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx)
    if is_method(m, :Basic, :GetEmpty)
        put!(chan.chan_get, GET_EMPTY_RESP)
    else
        msg = Message(UInt8[])
        if is_method(m, :Basic, :Deliver)
            msg.consumer_tag = m.payload.fields[1].second
            msg.delivery_tag = m.payload.fields[2].second
            msg.redelivered = convert(Bool, m.payload.fields[3].second)
            msg.exchange = convert(String, m.payload.fields[4].second)
            msg.routing_key = convert(String, m.payload.fields[5].second)
        else
            msg = Message(UInt8[])
            msg.delivery_tag = m.payload.fields[1].second
            msg.redelivered = convert(Bool, m.payload.fields[2].second)
            msg.exchange = convert(String, m.payload.fields[3].second)
            msg.routing_key = convert(String, m.payload.fields[4].second)
            msg.remaining = m.payload.fields[5].second
        end

        # wait for message header and body
        push!(chan.partial_msgs, msg)
    end
    nothing
end

function on_channel_message_in(chan::MessageChannel, m::TAMQPContentHeaderFrame, ctx)
    msg = last(chan.partial_msgs)
    msg.properties = m.hdrpayload.proplist
    msg.data = Vector{UInt8}(undef, m.hdrpayload.bodysize)
    msg.filled = 0
    nothing
end

function on_channel_message_in(chan::MessageChannel, m::TAMQPContentBodyFrame, ctx)
    msg = last(chan.partial_msgs)
    data = m.payload.data
    startpos = msg.filled + 1
    endpos = min(length(msg.data), msg.filled + length(data))
    msg.data[startpos:endpos] = data
    msg.filled = endpos

    if msg.filled >= length(msg.data)
        # got all data for msg
        if isempty(msg.consumer_tag)
            put!(chan.chan_get, pop!(chan.partial_msgs))
        else
            lock(chan.lck) do
                if msg.consumer_tag in keys(chan.consumers)
                    put!(chan.consumers[msg.consumer_tag].recvq, pop!(chan.partial_msgs))
                else
                    put!(get!(()->Channel{Message}(typemax(Int)), chan.pending_msgs, msg.consumer_tag), msg)
                    @debug("holding message, no consumer yet with tag", tag=msg.consumer_tag)
                end
            end
        end
    end

    nothing
end

on_confirm_select_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx) = _on_ack(chan, m, :Confirm, :SelectOk, ctx)

# ----------------------------------------
# send and recv for methods end
# ----------------------------------------


text:
using Statistics
using Elefridge
using PyPlot
using StatsBase

N = 10000
m = 10000
q = 7

meanerror = fill(0f0,3,q,N)
linerror = fill(0f0,3,q,N)
logerror = fill(0f0,3,q,N)

for i in 1:N
    # sample various distributions
    for (s,D) = enumerate([ rand(Float32,m),
                            4*rand(Float32,m),
                            rand(Float32,m).+1,
                            randn(Float32,m),
                            4*randn(Float32,m),
                            randn(Float32,m).+1,
                            exp.(randn(Float32,m))])

        Dround = round(D,7)
        Dgroom = groom(D,7)
        Dshave = shave(D,7)

        # mean error
        meanerror[1,s,i] = abs(mean(D - Dround))
        meanerror[2,s,i] = abs(mean(D - Dgroom))
        meanerror[3,s,i] = abs(mean(D - Dshave))

        # max lin error
        linerror[1,s,i] = maximum(abs.(D - Dround))
        linerror[2,s,i] = maximum(abs.(D - Dgroom))
        linerror[3,s,i] = maximum(abs.(D - Dshave))

        # max log error
        logerror[1,s,i] = maximum(abs.(log10.(D./Dround)))
        logerror[2,s,i] = maximum(abs.(log10.(D./Dgroom)))
        logerror[3,s,i] = maximum(abs.(log10.(D./Dshave)))
    end
 end

## fit histogram
ion()
pygui(true)
fig,(ax1,ax2,ax3) = subplots(1,3,figsize=(10,4),sharey=true)

lwmm = 1
lw10 = 3
lw25 = 8

distr = [   "Uniform(0,1)","Uniform(0,4)","Uniform(1,2)",
            "Normal(0,1)","Normal(0,4)","Normal(1,1)",
            "Lognormal(0,1)"]

for (ax,error) in zip([ax1,ax2,ax3],[meanerror,linerror,logerror])
    for i in 1:q
        for (s,e) in enumerate([error[1,i,:],error[2,i,:],error[3,i,:]])
            y = i-s/4
            c = "C"*string(s-1)
            ax.semilogx([minimum(e),maximum(e)],[y,y],c,lw=lwmm)
            p10 = percentile(e,10)
            p25 = percentile(e,25)
            p50 = percentile(e,50)
            p75 = percentile(e,75)
            p90 = percentile(e,90)
            ax.semilogx([p10,p90],[y,y],c,lw=lw10)
            ax.semilogx([p25,p75],[y,y],c,lw=lw25)
            ax.scatter(p50,y,100,c,edgecolor="k",zorder=5)
            ax.axhline(i,color="k",lw=0.5)
        end
    end
end

ax1.set_yticks([1,2,3,4,5,6,7].-0.5)
ax1.yaxis.set_tick_params(length=0)
ax2.yaxis.set_tick_params(length=0)
ax3.yaxis.set_tick_params(length=0)
ax1.set_yticklabels(distr)
ax1.set_ylim(-1.75,7)

ax1.set_xlim(1e-6,1e-2)
ax2.set_xlim(1e-3,1e0)
ax3.set_xlim(1e-3,1e0)

e = meanerror[2,1,:]
p10 = percentile(e,10)
p25 = percentile(e,25)
p50 = percentile(e,50)
p75 = percentile(e,75)
p90 = percentile(e,90)
maxx = maximum(e)
ax1.text(p10,0.1,"10%",rotation=90,ha="center",va="top",fontsize=9)
ax1.text(p25,0.1,"25%",rotation=90,ha="center",va="top",fontsize=9)
ax1.text(p50,0.1,"median",rotation=90,ha="center",va="top",fontsize=9)
ax1.text(p75,0.1,"75%",rotation=90,ha="center",va="top",fontsize=9)
ax1.text(p90,0.1,"90%",rotation=90,ha="center",va="top",fontsize=9)
ax1.text(maxx,0.1,"max",rotation=90,ha="center",va="top",fontsize=9)

ax2.plot(0,0,"C0",lw=8,label="Round-to-nearest")
ax2.plot(0,0,"C1",lw=8,label="Bit-grooming")
ax2.plot(0,0,"C2",lw=8,label="Bit-shaving")
ax2.legend(loc=4)

ax1.set_xlabel("absolute mean error")
ax2.set_xlabel("maximum absolute error")
ax3.set_xlabel("maximum decimal error")

ax1.set_title("Mean error",loc="left")
ax2.set_title("Absolute error",loc="left")
ax3.set_title("Decimal error",loc="left")

ax1.set_title("a",loc="right",fontweight="bold")
ax2.set_title("b",loc="right",fontweight="bold")
ax3.set_title("c",loc="right",fontweight="bold")

tight_layout()

ext: jl


text:
function _get_raster(::Type{CHELSA}, ::Type{BioClim}, layer::Integer)
    1 ≤ layer ≤ 19 || throw(ArgumentError("The layer must be between 1 and 19"))

    path = joinpath(SimpleSDMLayers._layers_assets_path, _rasterpath(CHELSA), _rasterpath(BioClim))
    isdir(path) || mkpath(path)

    layer = lpad(layer, 2, "0")
    filename = "CHELSA_bio10_$(layer).tif"
    url_root = "https://envicloud.os.zhdk.cloud.switch.ch/chelsa/chelsa_V1/climatologies/bio/"

    filepath = joinpath(path, filename)
    if !(isfile(filepath))
        Downloads.download(url_root * filename, filepath)
    end
    return filepath
end

function _get_raster(::Type{CHELSA}, ::Type{BioClim}, mod::CMIP5, fut::RepresentativeConcentrationPathway, layer::Integer, year="2041-2060")
    1 ≤ layer ≤ 19 || throw(ArgumentError("The layer must be between 1 and 19"))
    year in ["2041-2060", "2061-2080"] || throw(ArgumentError("The year must be 2041-2060 or 2061-2080"))

    path = joinpath(SimpleSDMLayers._layers_assets_path, _rasterpath(CHELSA), _rasterpath(BioClim), _rasterpath(mod), _rasterpath(fut), year)
    isdir(path) || mkpath(path)

    root = "https://os.zhdk.cloud.switch.ch/envicloud/chelsa/chelsa_V1/cmip5/$(year)/bio/"
    filename = "CHELSA_bio_mon_$(_rasterpath(mod))_$(_rasterpath(fut))_r1i1p1_g025.nc_$(layer)_$(year)_V1.2.tif"

    filepath = joinpath(path, filename)
    if !(isfile(filepath))
        Downloads.download(root * filename, filepath)
    end

    return filepath
end

function _get_raster(::Type{CHELSA}, ::Type{BioClim}, mod::CMIP6, fut::SharedSocioeconomicPathway, layer::Integer, year="2041-2060")
    1 ≤ layer ≤ 19 || throw(ArgumentError("The layer must be between 1 and 19"))
    year in ["2011-2040", "2041-2070", "2071-2100"] || throw(ArgumentError("The year must be 2011-2040, 2041-2070, or 2071-2100"))

    path = joinpath(SimpleSDMLayers._layers_assets_path, _rasterpath(CHELSA), _rasterpath(BioClim), _rasterpath(mod), _rasterpath(fut), year)
    isdir(path) || mkpath(path)

    root = "https://os.zhdk.cloud.switch.ch/envicloud/chelsa/chelsa_V2/GLOBAL/climatologies/$(year)/$(_rasterpath(mod))/$(_rasterpath(fut))/bio/"
    filename = "CHELSA_bio$(layer)_$(year)_$(lowercase(_rasterpath(mod)))_$(_rasterpath(fut))_V.2.1.tif"

    filepath = joinpath(path, filename)
    if !(isfile(filepath))
        Downloads.download(root * filename, filepath)
    end

    return filepath
end


text:
using PyPlot
using QuantEcon

ϕ_1, ϕ_2, ϕ_3, ϕ_4 = 0.5, -0.2, 0, 0.5
σ = 0.1

A = [ϕ_1 ϕ_2 ϕ_3 ϕ_4
     1.0 0.0 0.0 0.0
     0.0 1.0 0.0 0.0
     0.0 0.0 1.0 0.0]

C = [σ 0.0 0.0 0.0]'
G = [1.0 0.0 0.0 0.0]

T = 30
ar = LSS(A, C, G, mu_0=ones(4))

ymin, ymax = -0.8, 1.25

fix, axes = subplots(1, 2)

for ax in axes
    ax[:grid](alpha=0.4)
end

ax = axes[1]
ax[:set_ylim](ymin, ymax)
ax[:set_ylabel](L"$y_t$", fontsize=16)
ax[:vlines]((T,), -1.5, 1.5)
ax[:set_xticks]((T,))
ax[:set_xticklabels]((L"$T$",))

sample = {}

colors = ["c", "g", "b", "k"]

for i=1:20
    rcolor = colors[rand(1:3)]
    x, y = simulate(ar, T+15)
    y = squeeze(y, 1)
    ax[:plot](y, color=rcolor, lw=1, alpha=0.5)
    ax[:plot]((T,), (y[T], ), "ko", alpha=0.5)
    push!(sample, y[T])
end

axes[2][:set_ylim](ymin, ymax)
axes[2][:hist](sample, bins=16, normed=true, orientation="horizontal",
               alpha=0.5)


text:
const Δt_sec = 86400.0
const ldd_mv = 255

# read the staticmaps into memory
nc = NCDataset(staticmaps_rhine_path)
# helper function to get the axis order and directionality right
read_right(nc, var) = reverse(permutedims(nc[var][:]); dims = 2)
ldd_2d = read_right(nc, "ldd")

inds, _ = Wflow.active_indices(ldd_2d, ldd_mv)
n = length(inds)

# take out only the active cells
ldd = ldd_2d[inds]
slope = read_right(nc, "slope")[inds]
N = read_right(nc, "N")[inds]
Qold = read_right(nc, "Qold")[inds]
Bw = read_right(nc, "Bw")[inds]
waterlevel = read_right(nc, "waterlevel")[inds]
DCL = read_right(nc, "DCL")[inds]
close(nc)

# create the directed acyclic graph from the drainage direction array
graph = Wflow.flowgraph(ldd, inds, Wflow.pcr_dir)
# a topological sort is used for visiting nodes in order from upstream to downstream
toposort = topological_sort_by_dfs(graph)
sink = toposort[end]
@test ldd[sink] == 5  # the most downstream node must be a sink

# calculate parameters of kinematic wave
const q = 0.000001
const β = 0.6
const AlpPow = (2.0 / 3.0) * β
AlpTermR = (N ./ sqrt.(slope)) .^ β
P = Bw + (2.0 * waterlevel)
α = AlpTermR .* P .^ AlpPow

Q = zeros(n)
Q = Wflow.kin_wave!(Q, graph, toposort, Qold, q, α, β, DCL, Δt_sec)

@testset "flow rate" begin
    @test sum(Q) ≈ 2.957806043289641e6
    @test Q[toposort[1]] ≈ 0.007260052312634069f0
    @test Q[toposort[n-100]] ≈ 3945.762718338739f0
    @test Q[sink] ≈ 4131.101474418251
end

@testset "kinematic wave subsurface flow" begin
    @test all(
        isapprox.(
            Wflow.kinematic_wave_ssf(
                210128378079.0733,
                215395179156.82645,
                1540.34273559,
                1.238,
                18021.0,
                0.25,
                0.346,
                0.0017669756,
                1800.0,
                1.0,
                1697.05 * 1000.0,
                1200.0 * 1000.0,
                2443723.716252628,
            ),
            (7.410313985168225e10, 1540.1496836278836, -0.0),
        ),
    )
end

@testset "accucapacity" begin
    # test based on a subset of the examples at
    # https://pcraster.geo.uu.nl/pcraster/4.3.0/documentation/pcraster_manual/sphinx/op_accucapacity.html#examples
    # of the node at (row 3, column 2) and upstream nodes
    g = DiGraph(6)
    add_edge!(g, 1, 4)
    add_edge!(g, 2, 5)
    add_edge!(g, 3, 5)
    add_edge!(g, 4, 6)
    add_edge!(g, 5, 6)
    network = (graph = g, order = [1, 2, 3, 4, 5, 6])

    # example 1, accucapacityflux
    material = Float64[0.5, 2, 2, 0.5, 2, 0.5]
    capacity = fill(1.5, 6)
    flux, new_material = Wflow.accucapacityflux(material, network, capacity)
    @test new_material != material
    @test new_material == [0.0, 0.5, 0.5, 0.0, 3.5, 1.5]
    @test flux == Float64[0.5, 1.5, 1.5, 1, 1.5, 1.5]

    # example 2, accucapacityflux
    material = fill(10.0, 6)
    capacity = Float64[2, 30, 30, 2, 30, 2]
    flux, new_material = Wflow.accucapacityflux(material, network, capacity)
    @test new_material != material
    @test new_material == [8.0, 0.0, 0.0, 10.0, 0.0, 40.0]
    @test flux == Float64[2, 10, 10, 2, 30, 2]

    # example 1, accucapacitystate
    material = Float64[0.5, 2, 2, 0.5, 2, 0.5]
    capacity = fill(1.5, 6)
    new_material = Wflow.accucapacitystate(material, network, capacity)
    @test new_material != material
    @test new_material == [0.0, 0.5, 0.5, 0.0, 3.5, 1.5]

    # example 2, accucapacitystate
    material = fill(10.0, 6)
    capacity = Float64[2, 30, 30, 2, 30, 2]
    new_material = Wflow.accucapacitystate(material, network, capacity)
    @test new_material != material
    @test new_material == [8.0, 0.0, 0.0, 10.0, 0.0, 40.0]
end


text:
# Convert
@testset "Convert and constructors" begin
    @test Segment(1:5) == Segment(1, 5)
    @test Segment(4.2:9.2) == Segment(4.2, 9.2)
    @test first(SegmentValue(1:5, "Hi!")) == 1
    @test last(SegmentValue(1:5, "Hi!")) == 5
    @test value(SegmentValue(1:5, "Hi!")) == "Hi!"
    @test first(SegmentValue(4.2:9.2, "Bye!")) == 4.2
    @test last(SegmentValue(4.2:9.2, "Bye!")) == 9.2
    @test value(SegmentValue(4.2:9.2, "Bye!")) == "Bye!"
end

# Getters
@testset "Getters" begin
    i = Segment(5, 6)
    iv = SegmentValue(10, 11, "FOO")
    @test first(i)  === 5
    @test last(i)   === 6
    @test first(iv) === 10
    @test last(iv)  === 11
    @test value(iv) == "FOO"
end

# Generating random Segments
seed!(12345)
global maxend = round(Int, 1e9)

function randsegment(minstart=1, maxend=maxend)
    a = rand(minstart:maxend)
    b = rand(a:maxend)
    return (a, b)
end

# Verify that internal node keys and maxend values are correct
function validkeys(t::SegmentTrees.SegmentBTree{K, V, B}) where {K, V, B}
    minint = SegmentTrees.Segment{K}(0, 0)
    maxint = SegmentTrees.Segment{K}(maxend, maxend)
    return validkeys(t.root, minint, maxint)
end

function validkeys(node::SegmentTrees.InternalNode, minint, maxint)
    for (i, child) in enumerate(node.children)
        if child.maxend > node.maxend
            return false
        end

        if child.maxend != node.maxends[i]
            return false
        end
    end

    for i in 1:length(node.keys)
        k = node.keys[i]
        if SegmentTrees.minkey(node.children[i+1]) != k
            return false
        end

        if !validkeys(node.children[i], minint, k) ||
           !validkeys(node.children[i+1], k, maxint)
           return false
       end
    end
    return true
end

function validkeys(node::SegmentTrees.LeafNode, minint, maxint)
    for (key, entry) in zip(node.keys[1:node.count], node.entries[1:node.count])
        if first(key) != first(entry) || last(key) != last(entry)
            return false
        end

        segment = SegmentTrees.Segment(first(entry), last(entry))

        if !(minint <= segment <= maxint) || last(segment) > node.maxend
            return false
        end
    end
    return true
end

# Verify that parent pointers are correct
function validparents(t::SegmentTrees.SegmentBTree)
    if (t.root.parent !== nothing)
        return false
    end

    return validparents(t.root)
end

function validparents(node::SegmentTrees.InternalNode)
    for child in node.children
        if !validparents(child) || child.parent != node
            return false
        end
    end
    return true
end

validparents(node::SegmentTrees.LeafNode) = true

# Verify that sibling/cousin pointers are correct
function validsiblings(t::SegmentTrees.SegmentBTree)
    t.root.left === t.root.right === nothing || return false

    # Do an in-order traversal, pushing nodes onto a stack indexed by
    # depth.
    tdepth = depth(t)
    nodestacks = [Vector{SegmentTrees.Node}(undef, 0) for _ in 1:tdepth]

    function _visit(node::SegmentTrees.InternalNode, k)
        push!(nodestacks[k], node)
        for child in node.children
            _visit(child, k+1)
        end
    end

    function _visit(node::SegmentTrees.LeafNode, k)
        push!(nodestacks[k], node)
    end

    _visit(t.root, 1)

    for k in 1:tdepth
        stack = nodestacks[k]
        if isempty(stack)
            continue
        end

        if !(stack[1].left === nothing) || !(stack[end].right === nothing)
            return false
        end

        if length(stack) > 1
            if (stack[1].right === nothing)  || stack[1].right != stack[2] ||
               (stack[end].left === nothing) || stack[end].left != stack[end-1]
                return false
            end
        end

        for i in 2:length(stack)-1
            if (stack[i].left === nothing)  || stack[i].left != stack[i-1] ||
               (stack[i].right === nothing) || stack[i].right != stack[i+1]
                return false
            end
        end
    end

    return true
end


text:
using PyTSP
using Test

@testset "PyTSP.jl" begin
    @testset "Basic TSP solver tests" begin
        #  random TSP instance
        n_cities = 20
        factor = 1e5
        x = rand(n_cities) .* factor
        y = rand(n_cities) .* factor

        # The Concorde TSL Solver via pyconcorde
        time_concorde = @elapsed tour_concorde, length_concorde = solve_TSP_Concorde(x, y, norm="EUC_2D")

        # The LKH Heuristic via elkai
        time_LKH = @elapsed tour_LKH, length_LKH = solve_TSP_LKH(x, y)
        gap = (length_LKH - length_concorde) / length_LKH * 100
        gap = round(gap * 100) / 100

        # Compare the solutions
        println("============ n_cities = $n_cities ======================")
        @show tour_concorde
        @show tour_LKH
        println("Concorde      Tour Length: $length_concorde ($time_concorde seconds)")
        println("LKH           Tour Length: $length_LKH ($time_LKH seconds) (gap = $gap %)")

        @test isapprox(length_concorde, length_LKH; atol=2)
    end

    @testset "LKH type tests" begin
        #  random TSP instance
        n_cities = 20
        factor = 1e5
        x = rand(n_cities) .* factor
        y = rand(n_cities) .* factor

        tour_LKH_int1, length_LKH_int1 = solve_TSP_LKH(x, y)

        M = distance_matrix(x, y)
        tour_LKH_int2, length_LKH_int2 = solve_TSP_LKH(M)

        N = distance_matrix(x, y) .* 1.0
        tour_LKH_float, length_LKH_float = solve_TSP_LKH(N)

        @test isa(length_LKH_int1, Int)
        @test isa(length_LKH_int2, Int)
        @test isa(length_LKH_float, Float64)

        @test isapprox(length_LKH_int1, length_LKH_int2)
    end
end


text:
function pkg_fileinfo(id::PkgId)
    origin = get(Base.pkgorigins, id, nothing)
    origin === nothing && return nothing, nothing, nothing
    cachepath = origin.cachepath
    cachepath === nothing && return nothing, nothing, nothing
    provides, includes_requires, required_modules = try
        Base.parse_cache_header(cachepath; srcfiles_only=true)
    catch
        return nothing, nothing, nothing
    end
    includes, _ = includes_requires
    for (pkgid, buildid) in provides
        if pkgid.uuid === id.uuid && pkgid.name == id.name
            return cachepath, includes, first.(required_modules)
        end
    end
end

function parse_pkg_files(id::PkgId)
    pkgdata = get(pkgdatas, id, nothing)
    if pkgdata === nothing
        pkgdata = PkgData(id)
    end
    modsym = Symbol(id.name)
    if use_compiled_modules()
        cachefile, includes, reqs = pkg_fileinfo(id)
        if cachefile !== nothing
            @assert includes !== nothing
            @assert reqs !== nothing
            pkgdata.requirements = reqs
            for chi in includes
                mod = Base.root_module(id)
                for mpath in chi.modpath
                    mod = getfield(mod, Symbol(mpath))::Module
                end
                fname = relpath(chi.filename, pkgdata)
                # For precompiled packages, we can read the source later (whenever we need it)
                # from the *.ji cachefile.
                push!(pkgdata, fname=>FileInfo(mod, cachefile))
            end
            CodeTracking._pkgfiles[id] = pkgdata.info
            return pkgdata
        end
    end
    # Non-precompiled package(s). Here we rely on the `include` callbacks to have
    # already populated `included_files`; all we have to do is collect the relevant
    # files.
    # To reduce compiler latency, use runtime dispatch for `queue_includes!`.
    # `queue_includes!` requires compilation of the whole parsing/expression-splitting infrastructure,
    # and it's better to wait to compile it until we actually need it.
    Base.invokelatest(queue_includes!, pkgdata, id)
    return pkgdata
end

function modulefiles(mod::Module)
    function keypath(filename)
        filename = fixpath(filename)
        return get(src_file_key, filename, filename)
    end
    parentfile = String(first(methods(getfield(mod, :eval))).file)
    id = PkgId(mod)
    if id.name == "Base" || Symbol(id.name) ∈ stdlib_names
        parentfile = normpath(Base.find_source_file(parentfile))
        filedata = Base._included_files
        included_files = filter(mf->mf[1] == mod, filedata)
        return keypath(parentfile), [keypath(mf[2]) for mf in included_files]
    end
    use_compiled_modules() || return nothing, nothing   # FIXME: support non-precompiled packages
    _, filedata, reqs = pkg_fileinfo(id)
    filedata === nothing && return nothing, nothing
    included_files = filter(mf->mf.id == id, filedata)
    return keypath(parentfile), [keypath(mf.filename) for mf in included_files]
end


text:
# ------------------------------------------------------------------
# Licensed under the ISC License. See LICENSE in the project root.
# ------------------------------------------------------------------

"""
    uniquecoords(sdata, agg=Dict())

Retain locations in spatial objects with unique coordinates.

Duplicates of a variable `var` are aggregated with
aggregation function `agg[var]`. Default aggregation
function is `mean` for continuous variables and `first`
otherwise.
"""
function uniquecoords(sdata; agg=Dict())
  # retrieve filtering info
  vars = variables(sdata)
  for var in name.(vars)
    if var ∉ keys(agg)
      ST = scitype(sdata[var][1])
      agg[var] = ST <: Continuous ? _mean : _first
    end
  end

  # group locations with the same coordinates
  gind = _uniqueinds(coordinates(sdata), 2)
  groups = Dict(ind => Vector{Int}() for ind in unique(gind))
  for (i, ind) in enumerate(gind)
    push!(groups[ind], i)
  end

  # construct new point set data
  locs = Vector{Int}()
  vals = Dict(name(var) => Vector{mactype(var)}() for var in vars)
  for g in values(groups)
    i = g[1] # select any location
    if length(g) > 1
      # aggregate variables
      for var in name.(vars)
        push!(vals[var], agg[var](sdata[var][g]))
      end
    else
      # copy location
      for var in name.(vars)
        push!(vals[var], sdata[var][i])
      end
    end
    push!(locs, i)
  end

  # construct data with correct variable order
  ctor = Tables.materializer(values(sdata))
  data = ctor([var => vals[var] for var in name.(vars)])
  coords = coordinates(sdata, locs)

  georef(data, coords)
end

function _mean(xs)
  vs = skipmissing(xs)
  isempty(vs) ? missing : mean(vs)
end

function _first(xs)
  vs = skipmissing(xs)
  isempty(vs) ? missing : first(vs)
end

# ---------------------------------------------------------------
# The code below was copied/modified provisorily from Base.unique
# See https://github.com/JuliaLang/julia/issues/1845
# ---------------------------------------------------------------
using Base.Cartesian: @nref, @nloops

import Base: hash
struct Prehashed hash::UInt; end
hash(x::Prehashed) = x.hash

@generated function _uniqueinds(A::AbstractArray{T,N}, dim::Int) where {T,N}
  quote
    if !(1 <= dim <= $N)
      ArgumentError("Input argument dim must be 1 <= dim <= $N, but is currently $dim")
    end
    hashes = zeros(UInt, size(A, dim))

    # Compute hash for each row
    k = 0
    @nloops $N i A d->(if d == dim; k = i_d; end) begin
      @inbounds hashes[k] = hash(hashes[k], hash((@nref $N A i)))
    end

    # Collect index of first row for each hash
    uniquerow = Array{Int}(undef,size(A, dim))
    firstrow = Dict{Prehashed,Int}()
    for k = 1:size(A, dim)
      uniquerow[k] = get!(firstrow, Prehashed(hashes[k]), k)
    end
    uniquerows = collect(values(firstrow))

    # Check for collisions
    collided = falses(size(A, dim))
    @inbounds begin
      @nloops $N i A d->(if d == dim
                           k = i_d
                           j_d = uniquerow[k]
                         else
                           j_d = i_d
                         end) begin
        if (@nref $N A j) != (@nref $N A i)
          collided[k] = true
        end
      end
    end

    if any(collided)
      nowcollided = BitArray(undef, size(A, dim))
      while any(collided)
        # Collect index of first row for each collided hash
        empty!(firstrow)
        for j = 1:size(A, dim)
          collided[j] || continue
          uniquerow[j] = get!(firstrow, Prehashed(hashes[j]), j)
        end
        for v in values(firstrow)
          push!(uniquerows, v)
        end

        # Check for collisions
        fill!(nowcollided, false)
        @nloops $N i A d->begin
          if d == dim
            k = i_d
            j_d = uniquerow[k]
            (!collided[k] || j_d == k) && continue
          else
            j_d = i_d
          end
        end begin
          if (@nref $N A j) != (@nref $N A i)
            nowcollided[k] = true
          end
        end
        (collided, nowcollided) = (nowcollided, collided)
      end
    end

    ie = unique(uniquerow)
    ic_dict = Dict{Int,Int}()
    for k = 1:length(ie)
      ic_dict[ie[k]] = k
    end

    ic = similar(uniquerow)
    for k = 1:length(ic)
      ic[k] = ie[ic_dict[uniquerow[k]]]
    end

    ic
  end
end


text:
using Plots,DelimitedFiles,TriplotRecipes

f(x,y) = exp(0.1*sin(5.1*x + -6.2*y) + 0.3*cos(4.3*x + 3.4*y))

x,y = eachcol(readdlm(joinpath(@__DIR__,"dolphin.xy")))
t = readdlm(joinpath(@__DIR__,"dolphin.t"),Int)' .+ 1
z = f.(x,y)

plot(aspect_ratio=:equal,size=(800,720))
tripcolor!(x,y,z,t,color=:magma)
trimesh!(x,y,t,fillalpha=0.0,linecolor=:white)
savefig("dolphin.png")


text:
using Aerospace
using Base.Test

# write your own tests here
pitchi = 45.0*D2R;
yawi = 90.0*D2R;
rolli = 10.0*D2R;
Tr_bni = TR_BN(rolli, pitchi, yawi);

# Testing of the Algorithms
quat = QuatInit(rolli, pitchi, yawi);
Tr_bn = QuatToDCM(quat);
euler = ComputeEuler( Tr_bn );

# write your own tests here
@test (euler[1] -  rolli) < 1.0e-10;
@test (euler[2] -  pitchi) < 1.0e-10;
@test (euler[3] -  yawi) < 1.0e-10;
@test norm(Tr_bni - Tr_bn) < 1.0e-10;


text:
abstract type LoadingsPriorXMLElement <: MyXMLElement end

mutable struct IntegratedFactorsXMLElement <: ModelExtensionXMLElement
    el::XMLOrNothing
    precision_prior_el::XMLOrNothing
    loadings::MyXMLElement
    loadings_prior::MyXMLElement
    precision::MyXMLElement
    precision_scale::Float64
    precision_shape::Float64
    treeModel::TreeModelXMLElement
    tree_trait_ind::Int
    standardize_traits::Bool
    id::String
end

function IntegratedFactorsXMLElement(treeModel_el::TreeModelXMLElement,
                                     k::Int; trait_ind::Int = 1,
                                     orthonormal::Bool = false)

    p = treeModel_el.trait_dims[trait_ind]

    L = default_loadings(k, p)
    if orthonormal
        load_param = ScaledOrthogonalMatrix(L, "L", "scale", "U")
    else
        load_param = MatrixParameter(L, "L", ["L$i" for i = 1:size(L, 1)])
    end

    prior = NormalDistributionLikelihood(load_param, "L.prior")

    precision = Parameter(ones(p), bn.FACTOR_PRECISION, lower=0.0)
    precision_scale = 1.0
    precision_shape = 1.0
    return IntegratedFactorsXMLElement(nothing, nothing,
                                load_param,
                                prior,
                                precision, precision_scale,
                                precision_shape,
                                treeModel_el,
                                trait_ind, true,
                                bn.DEFAULT_IF_NAME)
end

function name(::IntegratedFactorsXMLElement)
    return bn.INTEGRATED_FACTORS
end

function get_trait_name(ifxml::IntegratedFactorsXMLElement)
    return ifxml.treeModel.node_traits[ifxml.tree_trait_ind]
end

function make_xmlelement(model::IntegratedFactorModel, tm::TreeModelXMLElement;
                         ind::Int = 1)
    ifx = IntegratedFactorsXMLElement(tm, tip_dimension(model), trait_ind = ind)
    set_loadings(ifx, model.L)
    set_precision(ifx, 1 ./ model.λ)
    return ifx
end

function copy(x::IntegratedFactorsXMLElement)
    return IntegratedFactorsXMLElement(
        x.el,
        x.precision_prior_el,
        x.loadings,
        x.loadings_prior,
        x.precision,
        x.precision_scale,
        x.precision_shape,
        x.treeModel,
        x.tree_trait_ind,
        x.standardize_traits,
        x.id
    )
end

function default_loadings(k::Int, p::Int)
    @assert k <= p
    L = zeros(k, p)
    for i = 1:k
        L[i, i] = 1.0
    end
    # L = randn(k, p)
    return L
end

function set_loadings!(ifxml::IntegratedFactorsXMLElement, L::S) where S <: AbstractArray{Float64, 2}
    set_mat!(ifxml.loadings, L)
end

function get_loadings_param(ifxml::IntegratedFactorsXMLElement)
    return ifxml.loadings
end

function get_precision(ifxml::IntegratedFactorsXMLElement)
    return ifxml.precision
end

function make_xml(ifxml::IntegratedFactorsXMLElement;
                    reference_precision::Bool = false)

    make_xml(ifxml.loadings)
    # ifxml.loadings_el = make_loadings(ifxml.loadings)
    actual_prior = nothing
    make_xml(ifxml.loadings_prior)

    el = new_element(bn.INTEGRATED_FACTORS)
    attrs = [(bn.ID, ifxml.id),
            (bn.TRAIT_NAME, get_trait_name(ifxml))]
    set_attributes(el, attrs)

    if ifxml.standardize_traits
        set_attribute(el, bn.STANDARDIZE, bn.TRUE)
    else
        set_attribute(el, bn.STANDARDIZE, bn.FALSE)
    end

    l_el = new_child(el, bn.LOADINGS)
    add_ref_el(l_el, ifxml.loadings.el)

    p_el = new_child(el, bn.PRECISION)
    if reference_precision
        add_ref_el(p_el, ifxml.precision)
    else
        prec_el = make_xml(ifxml.precision)
        add_child(p_el, prec_el)
    end
    add_ref_el(el, ifxml.treeModel)
    tp_el = new_child(el, bn.TRAIT_PARAMETER)
    tp_id = ifxml.treeModel.param_names[ifxml.tree_trait_ind]
    add_ref_el(tp_el, bn.PARAMETER, tp_id)

    ifxml.el = el

    if !reference_precision
        ifxml.precision_prior_el = make_precision_prior(ifxml)
    end

    return el
end

function make_precision_prior(ifxml::IntegratedFactorsXMLElement)
    el = new_element(bn.GAMMA_PRIOR)
    set_attributes(el, [bn.ID => "$(bn.FACTOR_PRECISION).prior",
                        bn.SCALE => string(ifxml.precision_scale),
                        bn.SHAPE => string(ifxml.precision_shape)])
    add_ref_el(el, ifxml.precision)
    return el
end

function get_hmc_parameter(ifxml::IntegratedFactorsXMLElement)
    make_xml(ifxml)
    if typeof(ifxml.loadings) <: ScaledOrthogonalMatrix
        return ifxml.loadings.U
    end
    return ifxml.loadings.el
end

function make_loadings(L::AbstractArray{Float64, 2})
    el = new_element(bn.MATRIX_PARAMETER)
    set_attribute(el, bn.ID, bn.DEFAULT_LOADINGS_ID)
    n, p = size(L)
    for i = 1:n
        add_parameter(el, id = "$(bn.DEFAULT_LOADINGS_ID)$i",
                        value = L[i, :])
    end
    return el
end

function get_priors(xml::IntegratedFactorsXMLElement)
    make_xml(xml)

    return [xml.precision_prior_el; get_priors(xml.loadings_prior)]
end

function get_loggables(xml::IntegratedFactorsXMLElement)
    make_xml(xml)

    loggables = [get_loggables(xml.loadings); xml.precision]
    loggables = [loggables; get_loggables(xml.loadings_prior)]

    return loggables
end

function get_precision_prior(xml::IntegratedFactorsXMLElement)
    return xml.precision_prior_el
end

function set_shrinkage_mults!(ifxml::IntegratedFactorsXMLElement;
                        shapes::Vector{Float64} = Float64[],
                        scales::Vector{Float64} = Float64[],
                        set_scale::Bool = true,
                        set_mults::Bool = true)
    # msls = ifxml.msls
    # if isnothing(msls)
    #     error("No shrinkage prior on the integrated factors model.")
    # else
    #     if length(shapes) > 0
    #         fill_shrinkage_array!(msls.shapes, shapes)
    #     end
    #     if length(scales) > 0
    #         fill_shrinkage_array!(msls.scales, scales)
    #     end
    # end
    scale = set_shrinkage_mults!(ifxml.loadings_prior, shapes=shapes,
                                 scales=scales,
                                 set_mults = set_mults)
    if set_scale
        set_scale!(ifxml.loadings, scale)
    end
end

function fill_shrinkage_array!(to_fill::Vector{Float64},
                               fill_from::Vector{Float64})
    n = length(to_fill)
    m = length(fill_from)
    if n == m
        to_fill .= fill_from
    elseif (n - 1) == m
        to_fill[2:end] .= fill_from
    else
        error("Not implemented for these array dimensions.")
    end
end

function set_loadings(pfa::IntegratedFactorsXMLElement,
                      L::AbstractArray{Float64, 2})
    mat_param = pfa.loadings
    set_mat!(mat_param, L)
end

function set_precision(ifa::IntegratedFactorsXMLElement,
                       mat::AbstractArray{Float64, 2})
    if !isdiag(mat)
        error("Factor model must have diagonal precision")
    end
    set_precision(ifa, diag(mat))
end

function set_precision(ifa::IntegratedFactorsXMLElement,
                                precs::AbstractArray{Float64, 1})
    p = length(precs)
    if length(ifa.precision) != p
        error("Cannot set precision. Incompatible dimensions.")
    end
    for i = 1:p
        if precs[i] <= 0
            error("Precisions must be positive.")
        end
    end
    set_value(ifa.precision, precs)
end

function get_loadings_scale(ifm::IntegratedFactorsXMLElement)
    loadings = ifm.loadings
    if typeof(loadings) <: ScaledOrthogonalMatrix
        return loadings.scale
    else
        error("Scale matrix does not exist " *
              "(loadings must be of type ScaledOrthogonalMatrix).")
    end
end

function get_loadings(ifm::IntegratedFactorsXMLElement)
    loadings = ifm.loadings
    # if typeof(loadings) <: ScaledOrthogonalMatrix
    #     return loadings.U
    # end
    return loadings
end


text:
module PNG
# Started as a fork of https://github.com/FugroRoames/LibPNG.jl

using libpng_jll
using FixedPointNumbers
using FileIO: DataFormat, @format_str, Stream, File, filename, stream
using ImageCore

libpng_wrap_dir = joinpath(@__DIR__, "..", "gen", "libpng")
using CEnum
include(joinpath(libpng_wrap_dir, "ctypes.jl"))
export Ctm, Ctime_t, Cclock_t
include(joinpath(libpng_wrap_dir, "libpng_common.jl"))
include(joinpath(libpng_wrap_dir, "libpng_api.jl"))

include("wraphelpers.jl")
include("io.jl")

end # module


text:
module Sirion

include("HDSim.jl")
include("IdealGas.jl")
include("CFL.jl")
include("Eulerian.jl")
include("Lagrangian.jl")
include("HLLC.jl")
include("RigidWall.jl")
include("FreeFlow.jl")
include("DifferentBC.jl")
include("SimpleFluxCalculator.jl")
include("Geometry.jl")
include("ZeroSource.jl")
include("SimpleCellUpdater.jl")
include("HydroSnapshot.jl")

end

max_stars_repo_name: bolverk/Sirion

max_issues_repo_name: bolverk/Sirion

max_forks_repo_name: bolverk/Sirion


text:
function reset_tdms()
  TDMS.flags = zero(UInt32)
  TDMS.nsos = zero(UInt32)
  TDMS.rdos = zero(UInt64)
  TDMS.n_ch = zero(UInt32)
  TDMS.ts = zero(Int64)
  TDMS.fs = zero(Float64)
  TDMS.ox = zero(Float64)
  TDMS.oy = zero(Float64)
  TDMS.oz = zero(Float64)
  TDMS.name = ""
  TDMS.hdr = Dict{String, Any}()

  return nothing
end


text:
#=

# 103: Boundary flux
([source code](SOURCE_URL))

We consider two test problems.

Testproblem A: Consider in $\Omega_1=(0,1)$
```math
- d_1 \Delta u_1  + k_1 u_1 = c_1
```
in  with homogeneous Neumann boundary conditons.

Testproblem B: Consider in $\Omega_2=(0,1) x (0, 1) $
```math
- d_2 \Delta u_2  + k_2 u_2 = c_2
```
in  with homogeneous Neumann boundary conditons
and at the right boundary, i.e. $ {1} x (0, 1) $
```math
- d_b \Delta v  + k_b v = c_b.
```
If d_1 = d_b, k_1 = k_b and c_1 = c_b, then u and v should coincide.
=#

module Example230_BoundaryFlux

using VoronoiFVM
using ExtendableGrids
using GridVisualize

function main(;n = 2 * 10, # n musst be an even number
              d1 = 5.0, db = 5.0, # prefactors (before diffusive part)
              kmax = 2.0, cmax = 3.0,
              Plotter = nothing,
              unknown_storage=:sparse)

    ###########################################################################
    ######################          1D problem           ######################
    ###########################################################################

    ispec_1D       = 1
    bulk_1D        = 1

    X              = range(0.0, stop = 1.0, length = n)
    length_x       = length(X)
    length_x_half  = Int(length_x/2)

    grid_1D        = simplexgrid(X)

    k1             = zeros(length_x);       c1 = zeros(length_x);

    k1[ 1:length_x_half ] .= kmax;  k1[ length_x_half+1:length_x ] .= 0.0  # prefactor before reactive part
    c1[ 1:length_x_half ] .= 0.0;   c1[ length_x_half+1:length_x ] .= cmax # source term

    #### discretization functions ####

    function flux!(f,u,edge)
        f[1] = d1 *  (u[1,1] - u[1,2])
    end

    function reaction!(f, u, node)
        f[1] = k1[node.index] *  u[1]
    end

    function source!(f, node::VoronoiFVM.Node)
        f[1] = c1[node.index]
    end

    sys_1D = VoronoiFVM.System(grid_1D, VoronoiFVM.Physics(flux  = flux!,  reaction  = reaction!, source = source!) )

    # enable species in only region
    enable_species!(sys_1D, ispec_1D, [bulk_1D])

    inival_1D  = unknowns(sys_1D);    inival_1D .= 0.0
    sol_1D     = unknowns(sys_1D)

    ## Create solver control info
    control   = VoronoiFVM.NewtonControl()

    ## Stationary solution of both problems
    solve!(sol_1D, inival_1D, sys_1D, control = control)

    p = GridVisualizer(;Plotter = Plotter, layout = (2,1), clear = true, resolution = (800,500))

    scalarplot!(p[1,1], grid_1D, sol_1D[1,:], show = true, title ="1D calculation (d1 = $d1, kmax = $kmax, cmax = $cmax)")

    ###########################################################################
    ######################          2D problem           ######################
    ###########################################################################

    grid_2D         = simplexgrid(X, X)

    ispec_2D        = 1
    ispec_boundary  = 2
    bulk_2D         = 1
    active_boundary = 2

    # parameters for the bulk problem
    d2              = 1.0
    k2              = 1.0
    c2              = 1.0

    #### discretization functions for bulk species ####
    function flux2D!(f,u,edge)
        f[ispec_2D] = d2 *  (u[ispec_2D, 1] - u[ispec_2D, 2])
    end

    function reaction2D!(f, u, node)
        f[ispec_2D] = k2 * u[ispec_2D]
    end

    function source2D!(f, node)
        f[ispec_2D] = c2
    end

    #### discretization functions for boundary species at active boundary ####
    function bflux!(f,u,bedge)
        if bedge.region == active_boundary
            f[ispec_boundary] = db * (u[ispec_boundary, 1] - u[ispec_boundary, 2])
        end
    end

    function breaction!(f, u, bnode)

        if bnode.region == active_boundary

             if bnode.coord[2, bnode.index] <= 0.5
                 kb = kmax
             else
                 kb = 0.0
             end

            f[ispec_boundary] = kb *  u[ispec_boundary]
        end
    end

    function bsource!(f, bnode)
        if bnode.region == active_boundary

            if bnode.coord[2, bnode.index] <= 0.5
                cb = 0.0
            else
                cb = cmax
            end

            f[ispec_boundary] = cb
        end
    end

    sys_2D    = VoronoiFVM.System(grid_2D, VoronoiFVM.Physics(flux = flux2D!, reaction = reaction2D!, source = source2D!,
                                                              bflux = bflux!, breaction = breaction!, bsource = bsource!),
                                  unknown_storage=unknown_storage)

    # enable species in only region
    enable_species!(         sys_2D, ispec_2D,       [bulk_2D]        )
    enable_boundary_species!(sys_2D, ispec_boundary, [active_boundary])

    inival_2D  = unknowns(sys_2D);        inival_2D .= 0.0
    sol_2D     = unknowns(sys_2D)

    solve!(sol_2D, inival_2D, sys_2D, control = control)

    # this is for variable transformation, since we consider right outer boundary and want to transform to x-axis.
    function tran32!(a,b)
        a[1] = b[2]
    end

    # note that if adjusting active_boundary to 3 or 4, then transform needs to be deleted.
    bgrid_2D  = subgrid(grid_2D, [active_boundary], boundary = true, transform = tran32!)
    sol_bound = view(sol_2D[ispec_boundary, :], bgrid_2D)

    scalarplot!(p[2,1], bgrid_2D, sol_bound, show = true, cellwise = true, title = "Active boundary in 2D (db = $db, kb = $kmax, cb = $cmax)")

    errorsol = VoronoiFVM.norm(sys_1D, sol_bound - sol_1D', 2)

    return errorsol

end # main

function test()
    main(unknown_storage=:dense)<1.0e-14 &&
    main(unknown_storage=:sparse)<1.0e-14
end

end # module


text:
function myfunction(n)
    for i = 1:n
        A = randn(100,100,20)
        m = maximum(A)
        Am = mapslices(sum, A; dims=2)
        B = A[:,:,5]
        Bsort = mapslices(sort, B; dims=1)
        b = rand(100)
        C = B.*b
    end
end

myfunction(1)

@profiler myfunction(10)

# Short for:
# Profile.clear()
# @profile myfunction(10)
# Juno.profiler()


text:
function u_test(x, y, σ)
    exp( - ( (x-π)^2 + (y-π)^2 ) / (2σ^2) )
end

function testTrapezoidalQuadrature()
    M = 64
    N = 64

    σ = 0.5

    grid = Grid2d(M,N)
    u    = get_field(grid)

    for i in 1:M
        for j in 1:N
            u[i,j]  = u_test(grid.x[i], grid.y[j], σ)
        end
    end

    uint = trapezoidal_quadrature(u, u, grid)

    @test uint - π*σ^2 ≈ zero(eltype(u)) atol=1E-14
end

testTrapezoidalQuadrature()