Pub/Sub

publish(channel, message)

Post a message to a channel.

subscribe(fn::Function,
          channel,
          channels...;
          stop_fn::Function=(msg) -> false,
          err_cb::Function=(err) -> rethrow(err))

Listen for messages published to the given channels in a do block. Optionally provide a stop function stop_fn(msg) which gets run as a callback everytime a subscription message is received, the subscription loop breaks if the stop_fn returns true. Optionally provide err_cb(err) function which gets run on encountering an exception in the main subscription loop.

Examples

julia> channels = ["first", "second"];

julia> publisher = Client();

julia> subscriber = Client();

julia> stop_fn(msg) = msg[end] == "close subscription";  # stop the subscription loop if the message matches

julia> messages = [];

julia> @async subscribe(channels...; stop_fn=stop_fn, client=subscriber) do msg
           push!(messages, msg)
       end;  # Without @async this function will block, alternatively use Thread.@spawn

julia> wait_until_subscribed(subscriber);

julia> subscriber.is_subscribed
true

julia> subscriber.subscriptions
Set{String} with 2 elements:
  "second"
  "first"

julia> publish("first", "hello"; client=publisher);

julia> publish("second", "world"; client=publisher);

julia> println(messages)
Any[["message", "first", "hello"], ["message", "second", "world"]]  # message has the format [<message type>, <channel>, <actual message>]

julia> unsubscribe("first"; client=subscriber);

julia> wait_until_channel_unsubscribed(subscriber, "first");

julia> subscriber.subscriptions
Set{String} with 1 element:
  "second"

julia> unsubscribe(; client=subscriber);  # unsubscribe from all channels

julia> wait_until_unsubscribed(subscriber);

julia> subscriber.is_subscribed
false

julia> subscriber.subscriptions
Set{String}()

unsubscribe([channels...]) -> nothing

Unsubscribes the client from the given channels, or from all of them if none is given.

psubscribe(fn::Function,
           pattern,
           patterns...;
           stop_fn::Function=(msg) -> false,
           err_cb::Function=(err) -> rethrow(err))

Listen for messages published to the given channels matching ghe given patterns in a do block. Optionally provide a stop function stop_fn(msg) which gets run as a callback everytime a subscription message is received, the subscription loop breaks if the stop_fn returns true. Optionally provide err_cb(err) function which gets run on encountering an exception in the main subscription loop.

Examples

julia> patterns = ["first*", "second*"];

julia> publisher = Client();

julia> subscriber = Client();

julia> stop_fn(msg) = msg[end] == "close subscription";  # stop the subscription loop if the message matches

julia> messages = [];

julia> @async psubscribe(patterns...; stop_fn=stop_fn, client=subscriber) do msg
           push!(messages, msg)
       end;  # Without @async this function will block, alternatively use Thread.@spawn

julia> wait_until_subscribed(subscriber);

julia> subscriber.is_subscribed
true

julia> subscriber.psubscriptions
Set{String} with 2 elements:
  "first*"
  "second*"

julia> publish("first_pattern", "hello"; client=publisher);

julia> publish("second_pattern", "world"; client=publisher);

julia> println(messages)
Any[["pmessage", "first*", "first_pattern", "hello"], ["pmessage", "second*", "second_pattern", "world"]]  # message has the format [<message type>, <pattern>, <channel>, <actual message>]

julia> punsubscribe("first*"; client=subscriber);

julia> wait_until_pattern_unsubscribed(subscriber, "first*");

julia> subscriber.psubscriptions
Set{String} with 1 element:
  "second*"

julia> punsubscribe(; client=subscriber);  # unsubscribe from all patterns

julia> wait_until_unsubscribed(subscriber);

julia> subscriber.is_subscribed
false

julia> subscriber.psubscriptions
Set{String}()

unsubscribe([channels...]) -> nothing

Unsubscribes the client from the given patterns, or from all of them if none is given.