Packages

combo_pubsub

0.2.0
0.2.0 0.1.0

Distributed Pub/Sub system and presence tracking system.

HexDocs

Current section

2 Versions

Jump to
Readme
2 Versions
0 Dependencies
1 Dependant
Files
Activity
Hide whitespace

Compare versions

6 files changed
+53 additions
-20 deletions
  @@ -1,7 +1,6 @@
1 - The MIT License (MIT)
1 + MIT License
2 2
3 3 Copyright (c) Zeke Dou
4 - Copyright (c) Chris McCord
5 4
6 5 Permission is hereby granted, free of charge, to any person obtaining a copy
7 6 of this software and associated documentation files (the "Software"), to deal
  @@ -1,9 +1,9 @@
1 1 {<<"links">>,
2 2 [{<<"Source">>,<<"https://github.com/combo-lab/combo_pubsub">>},
3 3 {<<"Changelog">>,
4 - <<"https://github.com/combo-lab/combo_pubsub/blob/v0.1.0/CHANGELOG.md">>}]}.
4 + <<"https://github.com/combo-lab/combo_pubsub/blob/v0.2.0/CHANGELOG.md">>}]}.
5 5 {<<"name">>,<<"combo_pubsub">>}.
6 - {<<"version">>,<<"0.1.0">>}.
6 + {<<"version">>,<<"0.2.0">>}.
7 7 {<<"description">>,
8 8 <<"Distributed Pub/Sub system and presence tracking system.">>}.
9 9 {<<"elixir">>,<<"~> 1.18">>}.
  @@ -172,6 +172,8 @@ defmodule Combo.PubSub do
172 172 ensure no messages are lost. See the "Safe Pool Size Migration" section
173 173 in the module documentation.
174 174 Defaults to the value of `:pool_size`.
175 + * `:dispatcher` - the default dispatcher module for broadcasts.
176 + Defaults to `Combo.PubSub`. Can be overridden at the function level.
175 177
176 178 """
177 179 @spec child_spec(keyword) :: Supervisor.child_spec()
  @@ -217,6 +219,30 @@ defmodule Combo.PubSub do
217 219 Registry.unregister(pubsub, topic)
218 220 end
219 221
222 + @doc """
223 + Unsubscribes the caller from the PubSub adapter's topic taking the metadata
224 + into consideration.
225 +
226 + Unlike `unsubscribe/2`, this function matches on the metadata provided as an
227 + option when subscribed. This is useful when you have multiple subscriptions for
228 + the same topic with different metadata.
229 +
230 + ## Example
231 +
232 + iex> PubSub.subscribe_match(:my_pubsub, "users:123", metadata: :fast)
233 + :ok
234 + iex> PubSub.subscribe_match(:my_pubsub, "users:123", metadata: :slow)
235 + :ok
236 + iex> PubSub.unsubscribe_match(:my_pubsub, "users:123", :fast)
237 + :ok
238 + # Only the :fast subscription is removed, :slow remains active
239 +
240 + """
241 + @spec unsubscribe_match(t, topic, term) :: :ok
242 + def unsubscribe_match(pubsub, topic, metadata) when is_atom(pubsub) and is_binary(topic) do
243 + Registry.unregister_match(pubsub, topic, metadata)
244 + end
245 +
220 246 @doc """
221 247 Broadcasts message on given topic across the whole cluster.
222 248
  @@ -230,9 +256,10 @@ defmodule Combo.PubSub do
230 256 See the "Custom dispatching" section in the module documentation.
231 257 """
232 258 @spec broadcast(t, topic, message, dispatcher) :: :ok | {:error, term}
233 - def broadcast(pubsub, topic, message, dispatcher \\ __MODULE__)
259 + def broadcast(pubsub, topic, message, dispatcher \\ nil)
234 260 when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
235 - {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
261 + {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
262 + dispatcher = dispatcher || default_dispatcher
236 263
237 264 with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
238 265 dispatch(pubsub, :none, topic, message, dispatcher)
  @@ -257,9 +284,10 @@ defmodule Combo.PubSub do
257 284 See the "Custom dispatching" section in the module documentation.
258 285 """
259 286 @spec broadcast_from(t, pid, topic, message, dispatcher) :: :ok | {:error, term}
260 - def broadcast_from(pubsub, from, topic, message, dispatcher \\ __MODULE__)
287 + def broadcast_from(pubsub, from, topic, message, dispatcher \\ nil)
261 288 when is_atom(pubsub) and is_pid(from) and is_binary(topic) and is_atom(dispatcher) do
262 - {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
289 + {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
290 + dispatcher = dispatcher || default_dispatcher
263 291
264 292 with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
265 293 dispatch(pubsub, from, topic, message, dispatcher)
  @@ -279,8 +307,10 @@ defmodule Combo.PubSub do
279 307 See the "Custom dispatching" section in the module documentation.
280 308 """
281 309 @spec local_broadcast(t, topic, message, dispatcher) :: :ok
282 - def local_broadcast(pubsub, topic, message, dispatcher \\ __MODULE__)
310 + def local_broadcast(pubsub, topic, message, dispatcher \\ nil)
283 311 when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
312 + {:ok, {_adapter, _name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
313 + dispatcher = dispatcher || default_dispatcher
284 314 dispatch(pubsub, :none, topic, message, dispatcher)
285 315 end
286 316
  @@ -301,8 +331,10 @@ defmodule Combo.PubSub do
301 331 See the "Custom dispatching" section in the module documentation.
302 332 """
303 333 @spec local_broadcast_from(t, pid, topic, message, dispatcher) :: :ok
304 - def local_broadcast_from(pubsub, from, topic, message, dispatcher \\ __MODULE__)
334 + def local_broadcast_from(pubsub, from, topic, message, dispatcher \\ nil)
305 335 when is_atom(pubsub) and is_pid(from) and is_binary(topic) and is_atom(dispatcher) do
336 + {:ok, {_adapter, _name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
337 + dispatcher = dispatcher || default_dispatcher
306 338 dispatch(pubsub, from, topic, message, dispatcher)
307 339 end
308 340
  @@ -323,9 +355,10 @@ defmodule Combo.PubSub do
323 355 See the "Custom dispatching" section in the module documentation.
324 356 """
325 357 @spec direct_broadcast(node_name, t, topic, message, dispatcher) :: :ok | {:error, term}
326 - def direct_broadcast(node_name, pubsub, topic, message, dispatcher \\ __MODULE__)
358 + def direct_broadcast(node_name, pubsub, topic, message, dispatcher \\ nil)
327 359 when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
328 - {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
360 + {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
361 + dispatcher = dispatcher || default_dispatcher
329 362 adapter.direct_broadcast(name, node_name, topic, message, dispatcher)
330 363 end
331 364
  @@ -333,7 +366,7 @@ defmodule Combo.PubSub do
333 366 Raising version of `broadcast/4`.
334 367 """
335 368 @spec broadcast!(t, topic, message, dispatcher) :: :ok
336 - def broadcast!(pubsub, topic, message, dispatcher \\ __MODULE__) do
369 + def broadcast!(pubsub, topic, message, dispatcher \\ nil) do
337 370 case broadcast(pubsub, topic, message, dispatcher) do
338 371 :ok -> :ok
339 372 {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
  @@ -344,7 +377,7 @@ defmodule Combo.PubSub do
344 377 Raising version of `broadcast_from/5`.
345 378 """
346 379 @spec broadcast_from!(t, pid, topic, message, dispatcher) :: :ok
347 - def broadcast_from!(pubsub, from, topic, message, dispatcher \\ __MODULE__) do
380 + def broadcast_from!(pubsub, from, topic, message, dispatcher \\ nil) do
348 381 case broadcast_from(pubsub, from, topic, message, dispatcher) do
349 382 :ok -> :ok
350 383 {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
  @@ -355,7 +388,7 @@ defmodule Combo.PubSub do
355 388 Raising version of `direct_broadcast/5`.
356 389 """
357 390 @spec direct_broadcast!(node_name, t, topic, message, dispatcher) :: :ok
358 - def direct_broadcast!(node_name, pubsub, topic, message, dispatcher \\ __MODULE__) do
391 + def direct_broadcast!(node_name, pubsub, topic, message, dispatcher \\ nil) do
359 392 case direct_broadcast(node_name, pubsub, topic, message, dispatcher) do
360 393 :ok -> :ok
361 394 {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
  @@ -367,7 +400,7 @@ defmodule Combo.PubSub do
367 400 """
368 401 @spec node_name(t) :: node_name
369 402 def node_name(pubsub) do
370 - {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
403 + {:ok, {adapter, name, _dispatcher}} = Registry.meta(pubsub, :pubsub)
371 404 adapter.node_name(name)
372 405 end
  @@ -25,8 +25,10 @@ defmodule Combo.PubSub.Supervisor do
25 25 opts[:registry_size] || opts[:pool_size] ||
26 26 System.schedulers_online() |> Kernel./(4) |> Float.ceil() |> trunc()
27 27
28 + dispatcher = Keyword.get(opts, :dispatcher, Combo.PubSub)
29 +
28 30 registry = [
29 - meta: [pubsub: {adapter, adapter_name}],
31 + meta: [pubsub: {adapter, adapter_name, dispatcher}],
30 32 partitions: partitions,
31 33 keys: :duplicate,
32 34 name: name
  @@ -96,7 +96,6 @@ defmodule Combo.Tracker do
96 96 """
97 97
98 98 use Supervisor
99 - require Logger
100 99 alias Combo.Tracker.Shard
101 100
102 101 @type presence :: {key :: String.t(), meta :: map}
Loading more files…