-module(superstream).

-export([declare/2,
         delete/2,
         declare_and_bind_partition/3,
         delete_partition/3]).

-include_lib("amqp_client/include/amqp_client.hrl").

declare(Channel, SuperStream) ->
    ArgumentsExchange = [{<<"x-super-stream">>, bool, true}],
    DeclareExchange =
        #'exchange.declare'{exchange = SuperStream,
                            type = <<"direct">>,
                            durable = true,
                            arguments = ArgumentsExchange},
    #'exchange.declare_ok'{} = amqp_channel:call(Channel, DeclareExchange),
    ok.

delete(Channel, SuperStream) ->
    DeleteExchange = #'exchange.delete'{exchange = SuperStream},
    #'exchange.delete_ok'{} = amqp_channel:call(Channel, DeleteExchange),
    ok.

declare_and_bind_partition(Channel, SuperStream, RoutingKey) ->
    Queue = queue_name(SuperStream, RoutingKey),
    PartitionIndex = 1,
    ArgumentsQueue =
        [{<<"x-queue-type">>, longstr, <<"stream">>},
         {<<"x-stream-partition-order">>, long, PartitionIndex},
         {<<"x-queue-leader-locator">>, longstr, <<"balanced">>}],
    DeclareQueue =
        #'queue.declare'{queue = Queue,
                         durable = true,
                         arguments = ArgumentsQueue},
    #'queue.declare_ok'{} = amqp_channel:call(Channel, DeclareQueue),

    Binding =
        #'queue.bind'{queue = Queue,
                      exchange = SuperStream,
                      routing_key = RoutingKey},
    #'queue.bind_ok'{} = amqp_channel:call(Channel, Binding),
    ok.

delete_partition(Channel, SuperStream, RoutingKey) ->
    Queue = queue_name(SuperStream, RoutingKey),
    DeleteQueue = #'queue.delete'{queue = Queue},
    #'queue.delete_ok'{} = amqp_channel:call(Channel, DeleteQueue),
    ok.

queue_name(SuperStream, RoutingKey) ->
    <<SuperStream/binary, "-", RoutingKey/binary>>.