diff --git a/channel-sender/config/config-local.yaml b/channel-sender/config/config-local.yaml index 5ecf5a82..7a569da6 100644 --- a/channel-sender/config/config-local.yaml +++ b/channel-sender/config/config-local.yaml @@ -61,6 +61,16 @@ channel_sender_ex: on_clean_close: 300 on_disconnection: 900 + # Defines when the channel process will be created. If 'true', the channel process will be + # created after calling the /create rest endpoint. When 'false', the channel process + # will be created only when the websocket or sse connection is established and credentials + # are exchanged. + # If your use case needs to route messages inmediatelly after exchanging credentials (/create + # endpoint), then it is recommended to set this parameter to 'true'. Otherwise, it is memory + # inneficient to create the channel process too early or even if the channel potentially + # is never going to be used. + create_channel_early: true + cowboy: # https://ninenines.eu/docs/en/cowboy/2.12/manual/cowboy_http/ protocol_options: diff --git a/channel-sender/config/config-local1.yaml b/channel-sender/config/config-local1.yaml index 30a4cfbb..090c0c4f 100644 --- a/channel-sender/config/config-local1.yaml +++ b/channel-sender/config/config-local1.yaml @@ -61,6 +61,16 @@ channel_sender_ex: on_clean_close: 300 on_disconnection: 900 + # Defines when the channel process will be created. If 'true', the channel process will be + # created after calling the /create rest endpoint. When 'false', the channel process + # will be created only when the websocket or sse connection is established and credentials + # are exchanged. + # If your use case needs to route messages inmediatelly after exchanging credentials (/create + # endpoint), then it is recommended to set this parameter to 'true'. Otherwise, it is memory + # inneficient to create the channel process too early or even if the channel potentially + # is never going to be used. + create_channel_early: true + no_start: false topology: strategy: Elixir.Cluster.Strategy.Gossip # for local development diff --git a/channel-sender/config/config-local2.yaml b/channel-sender/config/config-local2.yaml index 996eca3b..5dd1cbd0 100644 --- a/channel-sender/config/config-local2.yaml +++ b/channel-sender/config/config-local2.yaml @@ -61,6 +61,16 @@ channel_sender_ex: on_clean_close: 300 on_disconnection: 900 + # Defines when the channel process will be created. If 'true', the channel process will be + # created after calling the /create rest endpoint. When 'false', the channel process + # will be created only when the websocket or sse connection is established and credentials + # are exchanged. + # If your use case needs to route messages inmediatelly after exchanging credentials (/create + # endpoint), then it is recommended to set this parameter to 'true'. Otherwise, it is memory + # inneficient to create the channel process too early or even if the channel potentially + # is never going to be used. + create_channel_early: true + no_start: false topology: strategy: Elixir.Cluster.Strategy.Gossip # for local development diff --git a/channel-sender/lib/channel_sender_ex/application_config.ex b/channel-sender/lib/channel_sender_ex/application_config.ex index 9ddab9f1..e16ecda0 100644 --- a/channel-sender/lib/channel_sender_ex/application_config.ex +++ b/channel-sender/lib/channel_sender_ex/application_config.ex @@ -109,6 +109,10 @@ defmodule ChannelSenderEx.ApplicationConfig do Application.put_env(:channel_sender_ex, :channel_shutdown_on_disconnection, Map.get(channel_wait_times, "on_disconnection", 300)) + Application.put_env(:channel_sender_ex, :create_channel_early, + Map.get(fetch(config, :channel_sender_ex), "create_channel_early", true) + ) + Application.put_env(:channel_sender_ex, :prometheus_port, Map.get(fetch(config, :channel_sender_ex), "prometheus_port", 9568) ) diff --git a/channel-sender/lib/channel_sender_ex/core/channel.ex b/channel-sender/lib/channel_sender_ex/core/channel.ex index 405d75ab..7a9d9d11 100644 --- a/channel-sender/lib/channel_sender_ex/core/channel.ex +++ b/channel-sender/lib/channel_sender_ex/core/channel.ex @@ -674,8 +674,8 @@ defmodule ChannelSenderEx.Core.Channel do defp calculate_refresh_token_timeout(:websocket) do token_validity = get_param(:max_age, 900) - min_timeout = token_validity / 6 - # 15% of the token life for websocket connections + # 15% of the token life for websocket connections, plus a random drift between 1 and min_disconnection_tolerance time + min_timeout = (token_validity / 6) + :rand.uniform(get_param(:min_disconnection_tolerance, 50)) round(min_timeout * 1000) end diff --git a/channel-sender/lib/channel_sender_ex/core/security/channel_authenticator.ex b/channel-sender/lib/channel_sender_ex/core/security/channel_authenticator.ex index 30b240c4..3783a6ed 100644 --- a/channel-sender/lib/channel_sender_ex/core/security/channel_authenticator.ex +++ b/channel-sender/lib/channel_sender_ex/core/security/channel_authenticator.ex @@ -4,6 +4,7 @@ defmodule ChannelSenderEx.Core.Security.ChannelAuthenticator do """ alias ChannelSenderEx.Core.ChannelIDGenerator alias ChannelSenderEx.Core.ChannelSupervisor + alias ChannelSenderEx.Core.RulesProvider @type application() :: String.t() @type user_ref() :: String.t() @@ -14,7 +15,7 @@ defmodule ChannelSenderEx.Core.Security.ChannelAuthenticator do @spec create_channel(application(), user_ref(), meta()) :: {channel_ref(), channel_secret()} def create_channel(application, user_ref, meta \\ []) do {channel_ref, _channel_secret} = credentials = create_channel_data_for(application, user_ref) - {:ok, _pid} = ChannelSupervisor.register_channel({channel_ref, application, user_ref, meta}) + if create_channel_early?(), do: ChannelSupervisor.register_channel({channel_ref, application, user_ref, meta}) credentials end @@ -36,4 +37,10 @@ defmodule ChannelSenderEx.Core.Security.ChannelAuthenticator do channel_secret = ChannelIDGenerator.generate_token(channel_ref, app_id, user_ref) {channel_ref, channel_secret} end + + defp create_channel_early? do + RulesProvider.get(:create_channel_early) + rescue + _ -> true + end end diff --git a/channel-sender/mix.exs b/channel-sender/mix.exs index 5ac99506..76e0a50d 100644 --- a/channel-sender/mix.exs +++ b/channel-sender/mix.exs @@ -4,7 +4,7 @@ defmodule ChannelSenderEx.MixProject do def project do [ app: :channel_sender_ex, - version: "0.2.5", + version: "0.2.6", elixir: "~> 1.16", start_permanent: Mix.env() == :prod, deps: deps(), @@ -36,19 +36,19 @@ defmodule ChannelSenderEx.MixProject do defp deps do [ # {:dep_from_hexpm, "~> 0.3.0"}, - {:benchee, "~> 0.13", only: [:dev, :benchee]}, + {:benchee, "~> 1.4.0", only: [:dev, :benchee]}, {:cowboy, "~> 2.8"}, {:cowlib, "~> 2.9", override: true}, {:plug_cowboy, "~> 2.0"}, {:elixir_uuid, "~> 1.2"}, - {:gen_state_machine, "~> 2.0"}, + {:gen_state_machine, "~> 3.0"}, {:jason, "~> 1.2"}, {:cors_plug, "~> 3.0"}, {:hackney, "~> 1.20.1", only: :test}, {:plug_crypto, "~> 2.1"}, {:stream_data, "~> 0.4", only: [:test]}, {:gun, "~> 1.3", only: [:test, :benchee]}, - {:libcluster, "~> 3.4.1"}, + {:libcluster, "~> 3.5"}, {:vapor, "~> 0.10.0"}, {:mock, "~> 0.3.0", only: :test}, {:credo, "~> 1.7", only: [:dev, :test], runtime: false}, @@ -58,9 +58,9 @@ defmodule ChannelSenderEx.MixProject do {:telemetry_poller, "~> 1.1"}, {:cowboy_telemetry, "~> 0.4.0"}, {:telemetry, "~> 1.3"}, - {:eflambe, "~> 0.3.0"}, - {:meck, "0.9.2"}, - {:observer_cli, "~> 1.8"} + # {:eflambe, "~> 0.3.0"}, + # {:meck, "0.9.2"}, + # {:observer_cli, "~> 1.8"} # {:dep_from_git, git: "https://github.com/elixir-lang/my_dep.git", tag: "0.1.0"} ] end diff --git a/channel-sender/mix.lock b/channel-sender/mix.lock index 291d3a14..a5bbe54f 100644 --- a/channel-sender/mix.lock +++ b/channel-sender/mix.lock @@ -1,5 +1,5 @@ %{ - "benchee": {:hex, :benchee, "0.99.0", "0efbfc31045ad2f75a48673bd1befa8a6a5855e93b8c3117aed7d7da8de65b71", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}], "hexpm", "672d8e9436471b7d5b77ca5be3ad69d065553e7ed8c5db29bb3d662378104618"}, + "benchee": {:hex, :benchee, "1.4.0", "9f1f96a30ac80bab94faad644b39a9031d5632e517416a8ab0a6b0ac4df124ce", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}, {:statistex, "~> 1.0", [hex: :statistex, repo: "hexpm", optional: false]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "299cd10dd8ce51c9ea3ddb74bb150f93d25e968f93e4c1fa31698a8e4fa5d715"}, "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"}, "cachex": {:hex, :cachex, "4.1.0", "f9a9123feac30df2b75260c9a74903c0bd7ec54f345f5f6f60c7830b81ead43a", [:mix], [{:eternal, "~> 1.2", [hex: :eternal, repo: "hexpm", optional: false]}, {:ex_hash_ring, "~> 6.0", [hex: :ex_hash_ring, repo: "hexpm", optional: false]}, {:jumper, "~> 1.0", [hex: :jumper, repo: "hexpm", optional: false]}, {:sleeplocks, "~> 1.1", [hex: :sleeplocks, repo: "hexpm", optional: false]}, {:unsafe, "~> 1.0", [hex: :unsafe, repo: "hexpm", optional: false]}], "hexpm", "77f9417a5549e424bdbfeea3f8060841384c39c0b5f563a8f62f1f2c2b6d2345"}, "certifi": {:hex, :certifi, "2.12.0", "2d1cca2ec95f59643862af91f001478c9863c2ac9cb6e2f89780bfd8de987329", [:rebar3], [], "hexpm", "ee68d85df22e554040cdb4be100f33873ac6051387baf6a8f6ce82272340ff1c"}, @@ -16,13 +16,13 @@ "eternal": {:hex, :eternal, "1.2.2", "d1641c86368de99375b98d183042dd6c2b234262b8d08dfd72b9eeaafc2a1abd", [:mix], [], "hexpm", "2c9fe32b9c3726703ba5e1d43a1d255a4f3f2d8f8f9bc19f094c7cb1a7a9e782"}, "ex_hash_ring": {:hex, :ex_hash_ring, "6.0.4", "bef9d2d796afbbe25ab5b5a7ed746e06b99c76604f558113c273466d52fa6d6b", [:mix], [], "hexpm", "89adabf31f7d3dfaa36802ce598ce918e9b5b33bae8909ac1a4d052e1e567d18"}, "file_system": {:hex, :file_system, "1.1.0", "08d232062284546c6c34426997dd7ef6ec9f8bbd090eb91780283c9016840e8f", [:mix], [], "hexpm", "bfcf81244f416871f2a2e15c1b515287faa5db9c6bcf290222206d120b3d43f6"}, - "gen_state_machine": {:hex, :gen_state_machine, "2.1.0", "a38b0e53fad812d29ec149f0d354da5d1bc0d7222c3711f3a0bd5aa608b42992", [:mix], [], "hexpm", "ae367038808db25cee2f2c4b8d0531522ea587c4995eb6f96ee73410a60fa06b"}, + "gen_state_machine": {:hex, :gen_state_machine, "3.0.0", "1e57f86a494e5c6b14137ebef26a7eb342b3b0070c7135f2d6768ed3f6b6cdff", [:mix], [], "hexpm", "0a59652574bebceb7309f6b749d2a41b45fdeda8dbb4da0791e355dd19f0ed15"}, "gun": {:hex, :gun, "1.3.3", "cf8b51beb36c22b9c8df1921e3f2bc4d2b1f68b49ad4fbc64e91875aa14e16b4", [:rebar3], [{:cowlib, "~> 2.7.0", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "3106ce167f9c9723f849e4fb54ea4a4d814e3996ae243a1c828b256e749041e0"}, "hackney": {:hex, :hackney, "1.20.1", "8d97aec62ddddd757d128bfd1df6c5861093419f8f7a4223823537bad5d064e2", [:rebar3], [{:certifi, "~> 2.12.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.4.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "fe9094e5f1a2a2c0a7d10918fee36bfec0ec2a979994cff8cfe8058cd9af38e3"}, "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"}, "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"}, "jumper": {:hex, :jumper, "1.0.2", "68cdcd84472a00ac596b4e6459a41b3062d4427cbd4f1e8c8793c5b54f1406a7", [:mix], [], "hexpm", "9b7782409021e01ab3c08270e26f36eb62976a38c1aa64b2eaf6348422f165e1"}, - "libcluster": {:hex, :libcluster, "3.4.1", "271d2da892763bbef53c2872036c936fe8b80111eb1feefb2d30a3bb15c9b4f6", [:mix], [{:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.3", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "1d568157f069c6afa70ec0d736704cf799734bdbb6343f0322af4a980301c853"}, + "libcluster": {:hex, :libcluster, "3.5.0", "5ee4cfde4bdf32b2fef271e33ce3241e89509f4344f6c6a8d4069937484866ba", [:mix], [{:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.3", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ebf6561fcedd765a4cd43b4b8c04b1c87f4177b5fb3cbdfe40a780499d72f743"}, "libring": {:hex, :libring, "1.7.0", "4f245d2f1476cd7ed8f03740f6431acba815401e40299208c7f5c640e1883bda", [:mix], [], "hexpm", "070e3593cb572e04f2c8470dd0c119bc1817a7a0a7f88229f43cf0345268ec42"}, "meck": {:hex, :meck, "0.9.2", "85ccbab053f1db86c7ca240e9fc718170ee5bda03810a6292b5306bf31bae5f5", [:rebar3], [], "hexpm", "81344f561357dc40a8344afa53767c32669153355b626ea9fcbc8da6b3045826"}, "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"}, @@ -39,6 +39,7 @@ "recon": {:hex, :recon, "2.5.6", "9052588e83bfedfd9b72e1034532aee2a5369d9d9343b61aeb7fbce761010741", [:mix, :rebar3], [], "hexpm", "96c6799792d735cc0f0fd0f86267e9d351e63339cbe03df9d162010cefc26bb0"}, "sleeplocks": {:hex, :sleeplocks, "1.1.3", "96a86460cc33b435c7310dbd27ec82ca2c1f24ae38e34f8edde97f756503441a", [:rebar3], [], "hexpm", "d3b3958552e6eb16f463921e70ae7c767519ef8f5be46d7696cc1ed649421321"}, "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"}, + "statistex": {:hex, :statistex, "1.0.0", "f3dc93f3c0c6c92e5f291704cf62b99b553253d7969e9a5fa713e5481cd858a5", [:mix], [], "hexpm", "ff9d8bee7035028ab4742ff52fc80a2aa35cece833cf5319009b52f1b5a86c27"}, "stream_data": {:hex, :stream_data, "0.6.0", "e87a9a79d7ec23d10ff83eb025141ef4915eeb09d4491f79e52f2562b73e5f47", [:mix], [], "hexpm", "b92b5031b650ca480ced047578f1d57ea6dd563f5b57464ad274718c9c29501c"}, "swarm": {:hex, :swarm, "3.4.0", "64f8b30055d74640d2186c66354b33b999438692a91be275bb89cdc7e401f448", [:mix], [{:gen_state_machine, "~> 2.0", [hex: :gen_state_machine, repo: "hexpm", optional: false]}, {:libring, "~> 1.0", [hex: :libring, repo: "hexpm", optional: false]}], "hexpm", "94884f84783fc1ba027aba8fe8a7dae4aad78c98e9f9c76667ec3471585c08c6"}, "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"}, diff --git a/channel-sender/test/channel_sender_ex/core/channel_test.exs b/channel-sender/test/channel_sender_ex/core/channel_test.exs index 3ffed2b6..829eaeed 100644 --- a/channel-sender/test/channel_sender_ex/core/channel_test.exs +++ b/channel-sender/test/channel_sender_ex/core/channel_test.exs @@ -151,13 +151,13 @@ defmodule ChannelSenderEx.Core.ChannelTest do # :sys.trace(pid, true) :ok = Channel.socket_connected(pid, self()) refute_receive {:deliver_msg, _from = {^pid, _}, {_, "", ":n_token", _token, _}}, 100 - assert_receive {:deliver_msg, _from = {^pid, ref}, {id, "", ":n_token", _token, _}}, 500 + assert_receive {:deliver_msg, _from = {^pid, ref}, {id, "", ":n_token", _token, _}}, 2000 Channel.notify_ack(pid, ref, id) refute_receive {:deliver_msg, _from = {^pid, _}, {_, "", ":n_token", _token, _}}, 100 - assert_receive {:deliver_msg, _from = {^pid, ref}, {id, "", ":n_token", _token, _}}, 500 + assert_receive {:deliver_msg, _from = {^pid, ref}, {id, "", ":n_token", _token, _}}, 2000 Channel.notify_ack(pid, ref, id) refute_receive {:deliver_msg, _from = {^pid, _}, {_, "", ":n_token", _token, _}}, 100 - assert_receive {:deliver_msg, _from = {^pid, ref}, {id, "", ":n_token", token, _}}, 500 + assert_receive {:deliver_msg, _from = {^pid, ref}, {id, "", ":n_token", token, _}}, 2000 Channel.notify_ack(pid, ref, id) assert {:ok, _app, _user} = ChannelIDGenerator.verify_token(channel, token)