phone

    • chevron_right

      Erlang Solutions: You’ve been curious about LiveView, but you haven’t gotten into it

      news.movim.eu / PlanetJabber · Thursday, 6 April, 2023 - 10:04 · 21 minutes

    As a backend developer, I’ve spent most of my programming career away from frontend development. Whether it’s React/Elm for the web or Swift/Kotlin for mobile, these are fields of knowledge that fall outside of what I usually work with.

    Nonetheless, I always wanted to have a tool at my disposal for building rich frontends. While the web seemed like the platform with the lowest bar of entry for this, the size of the Javascript ecosystem had become so vast that familiarizing oneself with it was no small task.

    This is why I got very excited when Chris McCord first showed LiveView to the world. Building interactive frontends, with no Javascript required? This sounded like it was made for all of us Elixir backend developers that were “frontend curious”.

    However, if you haven’t already jumped into it, you might be hesitant to start. After all: it’s often not just about learning LiveView as if you were writing a greenfield project, but about how you would add LiveView into that Phoenix app that you’re already working on.

    Therefore, throughout this guide, I’ll presume that you already have an existing project that you wish to integrate LiveView into. If you have the luxury of a clean slate, then other resources (such as the Programming Phoenix LiveView book, by Bruce A. Tate and Sophie DeBenedetto ) may be of more use.

    I hope that this article may serve you well as a starting point!

    Will it work for my use case?

    You might have some worries about whether LiveView is a technology that you can introduce to your application. After all: no team likes to adopt a technology that they later figure out does not suit their use case.

    There are some properties of LiveView which are inherent to the technology, and therefore must be considered:

    Offline mode

    The biggest question is whether you need an offline mode for your application. My guess is that you probably do not need it , but if you do, LiveView is not the technology for you. The reason for this is that LiveView is rendered on the backend , necessitating communication with it.

    Latency

    The second biggest question: do you expect the latency from your clients to the server to be high , and would it being high be a serious detriment to your application?

    As Chris McCord put it in his announcement blog post on the Dockyard blog :

    “Certain use cases and experiences demand zero-latency, as well as offline capabilities. This is where Javascript frameworks like React, Ember, etc., shine.”

    Almost every interaction with a LiveView interface will send a request to the server; while requests will have highly optimized payloads, if you expect the average round trip from client to server to be too many milliseconds, then the user experience will suffer. LiveView ships with tools for testing your application with increased latency, but if you already know that there’s a certain latency maximum that your clients must not but very likely would exceed, then LiveView may not be suitable.

    If these are not of concern to your use case, then let’s get going!

    What does it take for me to start?

    Phoenix setup

    First of all, you’ll want to have a recent version of Phoenix, and your code up-to-date. Following are upgrade guides for older projects:

    LiveView setup

    The next step is to install LiveView into your existing project. The LiveView documentation has a great section on the subject: Installing LiveView into an existing project .

    The guide is rather straight-forward, so I will not reiterate its contents here. The only comment I’ll add is that the section at the very end about adding a topbar is (as the documentation points out) optional. It should be said, however, that this is added by default in new LiveView projects, so if you want to have a setup that’s as close to a freshly generated project, you should include this.

    At this point, you should have everything ready for introducing your own LiveView code!

    Quick LiveView overview

    Before we get to the actual coding, let’s get at a quick overview of the life cycle of a LiveView page. Here’s a high-level overview:

    The first request made to a LiveView route will be a plain HTTP request. The router will invoke a LiveView module, which calls the mount/3 function and then the render/1 function. This will render a static page (SEO-friendly out-of-the-box, by the way!), with the required Javascript for LiveView to work. The page then opens a WebSocket connection between the client and the server.

    After the WebSocket connection has been established, we get into the LiveView life cycle:

    Note that mount/3 and render/1 will be called again, this time over the WebSocket connection. While this probably will not be something you need to worry about when writing your first LiveView pages, it might be of relevance to know that this is the case ( discussion about this can be read here ). If you have a very expensive function call to make, and you only want to do it once, consider using the connected?/1 function.

    After render/1 has been called a second time, we get into the LiveView loop: wait for events, send the events over the wire, change the state on the server, then send back the minimal required data for updating the page on the client.

    Let’s now see how we’ll need to change your code to get to this LiveView flow.

    Making things live

    Now you might be asking:

    “OK, so the basics have been set up. What are the bare minimum things to get a page to be live?”

    You’ll need to do the following things:

    1. Convert an existing route to a live one
    2. Convert the controller module into a live module
    3. Modify the templates
    4. Introduce liveness

    Let’s go over them, one by one:

    Bringing life to the dead

    Here’s a question I once had, that you might be wondering:

    If I’ve got a regular (“dead”) Phoenix route, can I just add something live to a portion of the page, on the existing “dead” route?

    Considering how LiveView works, I’d like to transform the question into two new (slightly different) questions:

    1. Can one preserve the current routes and controllers, having them execute live code?
    2. Can one express the live interactions in the dead controllers?

    The answer to the first question: yes, but generally you won’t . You won’t, because of the answer to the second question: no , you’ll need separate live modules to express the live interactions.

    This leads to an important point:

    If you want some part of a page to be live, then your whole page has to be live.

    Technically , you can have the route be something else than live (e.g. a get route), and you would then use Phoenix.LiveView.Controller.live_render/3 in a “dead” controller function to render a LiveView module. This does still mean, however, that the page (the logic and templates) will be defined by the live module. You’re not “adding something live to a portion of the dead page”, but rather delegating to a live module from a dead route; you’ll still have to migrate the logic and templates to the live module.

    Therefore, your live code will be in LiveView modules (instead of your current controller modules ), invoked by live routes. As a sidenote: while it’s not covered by this article, you’ll eventually group live routes with live_session/3 , enabling redirects between routes without full page reloads.

    Introducing a live route

    Many tutorials and videos about LiveView use the example of programming a continuously updating rendering of a thermostat. Let’s therefore presume that you’ve got a home automation application, and up until now you had to go to /thermostats and refresh the page to get the latest data.

    The router.ex might look something like this:

    defmodule HomeAutomationWeb.Router do
      use HomeAutomationWeb, :router
    
      pipeline :browser do
        # ...
      end
    
      pipeline :logged_in do
        # ...
      end
    
      scope "/", HomeAutomationWeb do
        pipe_through [:browser, :logged_in]
    
        # ...
    
        resources "/thermostats", ThermostatController
        post "/thermostats/reboot", ThermostatController, :reboot
      end
    end
    

    This is a rather simple router (with some lines removed for brevity), but you can probably figure out how this compares to your code. We’re using a call to Phoenix.Router.resources/2 here to cover a standard set of CRUD actions; your set of actions could be different.

    Let’s introduce the following route after the post-route:

    live "/live/thermostats", ThermostatLive
    

    The ThermostatLive will be the module to which we’ll be migrating logic from ThermostatController.

    Creating a live module to migrate to

    Creating a skeleton

    Let’s start by creating a directory for LiveView modules, then create an empty thermostat_live.ex in that directory.

    $ mkdir lib/home_automation_web/live
    $ touch lib/home_automation_web/live/thermostat_live.ex
    

    It might seem a bit strange to create a dedicated directory for the live modules, considering that the dead parts of your application already have controller/template/view directories. This convention, however, allows one to make use of the following feature from the Phoenix.LiveView.render/1 callback (slight changes by me, for readability):

    If you don’t define [render/1 in your LiveView module], LiveView will attempt to render a template in the same directory as your LiveView. For example, if you have a LiveView named MyApp.MyCustomView inside lib/my_app/live_views/my_custom_view.ex, Phoenix will look for a template at lib/my_app/live_views/my_custom_view.html.heex.

    This means that it’s common for LiveView projects to have a live directory with file pairs, such as foobar.ex and foobar.html.heex, i.e. module and corresponding template. Whether you inline your template in the render/1 function or put it in a dedicated file is up to you.

    Open the lib/home_automation_web/live/thermostat_live.ex file, and add the following skeleton of the ThermostatLive module:

    defmodule HomeAutomationWeb.ThermostatLive do
      use HomeAutomationWeb, :live_view
    
      def mount(_params, _session, socket) do
        {:ok, socket}
      end
    
      def render(assigns) do
        ~H"""
        <div id="thermostats">
          <p>Thermostats</p>
        </div>
        """
      end
    end
    

    There are two mandatory callbacks in a LiveView module: mount/3, and render/1. As mentioned earlier, you can leave out render/1 if you have a template file with the right file name. You can also leave out the mount/3, but that would mean that you neither want to set any state, nor do any work on mount, which is unlikely.

    Migrating mount logic

    Let’s now look at our imagined HomeAutomationWeb.ThermostatController, to see what we’ll be transferring over to ThermostatLive:

    defmodule HomeAutomationWeb.ThermostatController do
      use HomeAutomationWeb, :controller
    
      alias HomeAutomation.Thermostat
    
      def index(conn, _params) do
        thermostats = Thermostat.all_for_user(conn.assigns.current_user)
    
        render(conn, :index, thermostats: thermostats)
      end
    
      # ...
    
      def reboot(conn, %{"id" => id}) do
        {:ok, thermostat} =
          id
          |> Thermostat.get!()
          |> Thermostat.reboot()
    
        conn
        |> put_flash(:info, "Thermostat '#{thermostat.room_name}' rebooted.")
        |> redirect(to: Routes.thermostat_path(conn, :index))
      end
    end
    

    We’ll be porting a subset of the functions that are present in the controller module: index/2 and reboot/2. This is mostly to have two somewhat different controller actions to work with.

    Let’s first focus on the index/2 function. We could imagine that Thermostat.all_for_user/1 makes a database call of some kind, possibly with Ecto. conn.assigns.current_user would be added to the assigns by the logged_in Plug in the pipeline in the router.

    Let’s naively move over the ThermostatController.index/2 logic to the LiveView module, and take it from there:

    defmodule HomeAutomationWeb.ThermostatLive do
      use HomeAutomationWeb, :live_view
    
      alias HomeAutomation.Thermostat
    
      def mount(_params, _session, socket) do
        thermostats = Thermostat.all_for_user(socket.assigns.current_user)
    
        {:ok, assign(socket, %{thermostats: thermostats})}
      end
    
      def render(assigns) do
        ~H"""
        <div id="thermostats">
          <p>Thermostats</p>
        </div>
        """
      end
    end
    

    Firstly, we’re inserting the index/2 logic into the mount/3 function of ThermostatLive, meaning that the data will be called for on page load.

    Secondly, notice that we changed the argument to Thermostat.all_for_user/1 from conn.assigns.current_user to socket.assigns.current_user. This is just a change of variable name, of course, but it signifies a change in the underlying data structure: you’re not working with a Plug.Conn struct, but rather with a Phoenix.LiveView.Socket.

    So far we’ve written some sample template code inside the render/1 function definition, and we haven’t seen the actual templates that would render the thermostats, so let’s get to those.

    Creating live templates

    Let’s presume that you have a rather simple index page, listing all of your thermostats.

    <h1>Listing Thermostats</h1>
    
    <%= for thermostat <- @thermostats do %>
      <div class="thermostat">
        <div class="row">
          <div class="column">
            <ul>
              <li>Room name: <%= thermostat.room_name %></li>
              <li>Temperature: <%= thermostat.temperature %></li>
            </ul>
          </div>
    
          <div class="column">
            Actions: <%= link("Show", to: Routes.thermostat_path(@conn, :show, thermostat)) %>
            <%= link("Edit", to: Routes.thermostat_path(@conn, :edit, thermostat)) %>
            <%= link("Delete",
              to: Routes.thermostat_path(@conn, :delete, thermostat),
              method: :delete,
              data: [confirm: "Are you sure?"]
            ) %>
          </div>
    
          <div class="column">
            <%= form_for %{}, Routes.thermostat_path(@conn, :reboot), fn f -> %>
              <%= hidden_input(f, :id, value: thermostat.id) %>
              <%= submit("Reboot", class: "rounded-full") %>
            <% end %>
          </div>
        </div>
      </div>
    <% end %>
    
    <%= link("New Thermostat", to: Routes.thermostat_path(@conn, :new)) %>
    

    Each listed thermostat has the standard resource links of Show/Edit/Delete, with a New-link at the very end of the page. The only thing that goes beyond the usual CRUD actions is the form_for, defining a Reboot-button. The Reboot-button will initiate a request to the POST /thermostats/reboot route.

    As previously mentioned, we can either move this template code into the ThermostatLive.render/1 function, or we can create a template file named lib/home_automation_web/live/thermostat_live.html.heex. To get used to the new ways of LiveView, let’s put the code into the render/1 function. You can always extract it later (but remember to delete the render/1 function, if you do!).

    The first step would be to simply copy paste everything, with the small change that you need to replace every instance of @conn with @socket. Here’s what the ThermostatLive will look like:

    defmodule HomeAutomationWeb.ThermostatLive do
      use HomeAutomationWeb, :live_view
    
      alias HomeAutomation.Thermostat
    
      def mount(_params, _session, socket) do
        thermostats = Thermostat.all_for_user(socket.assigns.current_user)
    
        {:ok, assign(socket, %{thermostats: thermostats})}
      end
    
      def render(assigns) do
        ~H"""
        <h1>Listing Thermostats</h1>
    
        <%= for thermostat <- @thermostats do %>
          <div class="thermostat">
            <div class="row">
              <div class="column">
                <ul>
                  <li>Room name: <%= thermostat.room_name %></li>
                  <li>Temperature: <%= thermostat.temperature %></li>
                </ul>
              </div>
    
              <div class="column">
                Actions: <%= link("Show", to: Routes.thermostat_path(@socket, :show, thermostat)) %>
                <%= link("Edit", to: Routes.thermostat_path(@socket, :edit, thermostat)) %>
                <%= link("Delete",
                  to: Routes.thermostat_path(@socket, :delete, thermostat),
                  method: :delete,
                  data: [confirm: "Are you sure?"]
                ) %>
              </div>
    
              <div class="column">
                <%= form_for %{}, Routes.thermostat_path(@socket, :reboot), fn f -> %>
                  <%= hidden_input(f, :id, value: thermostat.id) %>
                  <%= submit("Reboot", class: "rounded-full") %>
                <% end %>
              </div>
            </div>
          </div>
        <% end %>
    
        <%= link("New Thermostat", to: Routes.thermostat_path(@socket, :new)) %>
        """
      end
    end
    

    While this makes the page render, both the links and the form are doing the same “dead” navigation as before, leading to full-page reloads, not to mention that we currently get out from the live page.

    To make the page more live, let’s focus on making the clicking of the Reboot-button result in a LiveView event, instead of a regular POST with subsequent redirect.

    Changing the button to something live

    The Reboot-button is a good target to turn live, as it should just fire an asynchronous event, without redirecting anywhere. Let’s have a look at how the button is currently defined:

    <%= form_for %{}, Routes.thermostat_path(@socket, :reboot), fn f -> %>
      <%= hidden_input(f, :id, value: thermostat.id) %>
      <%= submit("Reboot", class: "rounded-full") %>
    <% end %>
    

    The reason why the “dead” template used a form_for with a submit is two-fold. Firstly , since the action of rebooting the thermostat is not a navigation action, using an anchor tag (<a>) styled to look like a button would not be appropriate: using a form with a submit button is better, since it indicates that an action will be performed, and the action is clearly defined by the form’s method and action attributes. Secondly , a form allows you to include a CSRF token , which is automatically injected into the resulting <form> with form_for.

    Let’s look at what the live version will look like:

    <%= link("Reboot",
      to: "#",
      phx_click: "reboot",
      phx_value_id: thermostat.id,
      data: [confirm: "Are you sure?"]
    ) %>
    

    Let’s break this down a bit:

    A note about <form>

    First thing to note: this is no longer a <form>!

    Above I mentioned CSRF protection being a reason for using the <form>, but the Channel (i.e. the WebSocket connection between server and client) is already protected with a CSRF token, so we can send LiveView events without worrying about this.

    The detail above about navigation technically still applies, but in LiveView one would (generally) use a link with to: “#” for most things functioning like a button.

    As a minor note: you’ll still be using forms in LiveView for data input, although you’ll be using the <.form> component , instead of calling form_for .

    The phx_click event

    The second thing to note is that is the phx_click attribute, and it’s value “reboot”. The key is indicating what event should be fired when interacting with the generated <a> tag. The various possible event bindings can be found here:

    https://hexdocs.pm/phoenix_live_view/bindings.html

    If you want to have a reference for what events you can work with in LiveView, the link above is a good one to bookmark!

    Clarifying a potentially confusing detail: the events listed in the above linked documentation use hyphens (-) as separators in their names. link uses underscores (_), but apart from this, the event names are the same.

    The “reboot” string specifies the “name” of the event that is sent to the server. We’ll see the usage of this string in a second.

    The value attribute

    Finally, let’s talk about the phx_value_id attribute. phx_value_id is special, in that part of the attribute name is user defined. The phx_value_-part of the attribute name indicates to LiveView that the attribute is an “event value”, and what follows after phx_value_ (in our case: id) will be the key name in the resulting “event data map” on the server side. The value of the attribute will become the value in the map.

    This means that this…:

    phx_value_id: "thermostat_13" ,

    …will be received as the following on the server:

    %{id: "thermostat_13"}

    Further explanation can be found in the documentation:

    https://hexdocs.pm/phoenix_live_view/bindings.html#click-events

    Adding the corresponding event to the LiveView module

    Now that we’ve changed the Reboot-button in the template, we can get to the final step: amending the ThermostatLive module to react to the “reboot” event. We need to add a handle_event function to the module, and we’ll use the logic that we saw earlier in ThermostatController.reboot/2:

    defmodule HomeAutomationWeb.ThermostatLive do
      use HomeAutomationWeb, :live_view
    
      alias HomeAutomation.Thermostat
    
      def mount(_params, _session, socket) do
        # ...
      end
    
      def handle_event("reboot", %{"id" => id}, socket) do
        {:ok, thermostat} =
          id
          |> Thermostat.get!()
          |> Thermostat.reboot()
    
        {:noreply,
          put_flash(
            socket,
            :info,
            "Thermostat '#{thermostat.room_name}' rebooted."
          )}
      end
    
      def render(assigns) do
        # ...
      end
    end
    

    This handle_event function will react to the “reboot” event. The first argument to the function is the event name, the second is any passed data (through phx-value-*), and finally the socket.

    A quick note about the :noreply: presume that you’ll be using {:noreply, socket}, as the alternative ({:reply, map, socket}) is rarely useful. Just don’t worry about this, for now.

    That’s it!

    If you’ve been following this guide, trying to adapt it to your application, then you should have something like the following:

    1. A live route.
    2. A live module, where you’ve ported some of the logic from the controller module.
    3. A template that’s been adapted to be rendered by a live module.
    4. An element on the page that, when interacted with, causes an event to fire, with no need for a page refresh.

    At this stage, one would probably want to address the other CRUD actions, at the very least having their navigation point to the live route, e.g. creating a new thermostat should not result in a redirect to the dead route. Even better would be to have the CRUD actions all be changed to be fully live, requiring no page reloads. However, this is unfortunately outside of the scope of this guide.

    I hope that this guide has helped you to take your first steps toward working with LiveView!

    Further reading

    Here’s some closing advice that you might find useful, if you want to continue on your own.

    Exploring generators

    A very educative thing to do is comparing what code Phoenix generates for “dead” pages vs. live pages.

    Following are the commands for first generating a “dead” CRUD page setup for a context (Devices) and entity (Thermostat), and then one generates the same context and entity, but in a live fashion. The resulting git commits illustrate how the same intent is expressed in the two styles.

    $ mix phx.new home_automation --live
    $ cd home_automation
    $ git init .
    $ git add .
    $ git commit -m "Initial commit"
    $ mix phx.gen.html Devices Thermostat thermostats room_name:string temperature:integer
    $ git add .
    $ git commit -m "Added Devices context with Thermostat entity"
    $ git show
    $ mix phx.gen.live Devices Thermostat thermostats room_name:string temperature:integer
    $ git add .
    $ git commit -m "Added Live version of Devices with Thermostat"
    $ git show
    

    Note that when you get to the phx.gen.live step, you’ll have to answer Y to a couple of questions, as you’ll be overwriting some code. Also, you’ll generate a superfluous Ecto migration, which you can ignore.

    Study these generated commits, the resulting files, and the difference between the generated approaches, as it helps a lot with understanding how the transition from dead to live is done.

    Broadcasting events

    You might want your live module to react to specific events in your application. In the case of the thermostat application it could be the change of temperature on any of the thermostats, or the reboot status getting updated asynchronously. In the case of a LiveView chat application, it would be receiving a new message from someone in the conversation.

    A very commonly used method for generating and listening to events is making use of Phoenix.PubSub . Not only is Phoenix.PubSub a robust solution for broadcasting events, it gets pulled in as a dependency to Phoenix, so you should already have the hex installed.

    There are numerous guides out there for how to make use of Phoenix.PubSub, but a good place to start is probably watching how Chris McCord uses LiveView and Phoenix.PubSub to create a Twitter clone, in about 15 minutes (the part with Phoenix.PubSub is about half-way through the video).

    HTTP verbs

    Regarding HTTP verbs, coming from the world of dead routes, you might be wondering:

    I’ve got various GET/POST/PUT/etc. routes that serve different purposes. When building live modules, do all of the routes (with their different HTTP verbs) just get replaced with live?

    Yes, mostly. Generally your live parts of the application will handle their communication over the WebSocket connection, sending various events. This means that any kind of meaning you wish to communicate through the various HTTP verbs will instead be communicated through various events instead.

    With that said, you may still have parts of your application that will still be accessed with regular HTTP requests, which would be a reason to keep these routes around. The will not, however, be called from your live components.

    Credits

    Last year, Stone Filipczak wrote an excellent guide on the SmartLogic blog , on how to quickly introduce LiveView to an existing phoenix app. It was difficult to not have overlap with that guide, so my intention has been to complement it. Either way, I encourage you to check it out!

    The post You’ve been curious about LiveView, but you haven’t gotten into it appeared first on Erlang Solutions .

    • wifi_tethering open_in_new

      This post is public

      www.erlang-solutions.com /blog/youve-been-curious-about-liveview-but-you-havent-gotten-into-it/

    • chevron_right

      Erlang Solutions: Captura de datos con Postgres y Elixir

      news.movim.eu / PlanetJabber · Wednesday, 5 April, 2023 - 06:37 · 20 minutes

    La captura de datos es el proceso de identificar y capturar cambios de datos en la base de datos.

    Con captura de datos, los cambios en los datos pueden ser rastreados casi en tiempo real, y esa información puede ser utilizada para apoyar una variedad de casos de uso, incluyendo auditoría, replicación y sincronización.

    Un buen ejemplo de un caso de uso para captura de datos es considerar una aplicación que inserta un registro en la base de datos y envía un evento a una cola de mensajes después de que se ha insertado el registro (escribir dos veces).

    Imagina que estás trabajando en una aplicación de comercio electrónico y después de que se crea y se inserta un pedido en la base de datos, se envía un evento OrderCreated a una cola de mensajes. Los consumidores del evento podrían hacer cosas como crear órdenes de recolección para el almacén, programar transportes para la entrega y enviar un correo electrónico de confirmación del pedido al cliente.

    Pero ¿qué sucede si la aplicación se bloquea después de que se ha insertado el pedido en la base de datos pero antes de lograr enviar el evento a la cola de mensajes? Esto es posible debido al hecho de que no se puede insertar atómicamente el registro Y enviar el mensaje en la misma transacción, por lo que si la aplicación se bloquea después de insertar el registro en la base de datos pero antes de enviar el evento a la cola, se pierde el evento.

    Por supuesto, existen soluciones alternativas para evitar esto: una solución simple es “almacenar” el evento en una tabla de almacenamiento temporal en la misma transacción en la que se escribe el registro, y luego depender de un proceso captura de datos para capturar el cambio en la tabla de almacenamiento y enviar el evento a la cola de mensajes. La transacción es atómica y el proceso de captura de datos puede asegurar que el evento se entregue al menos una vez.

    Para capturar cambios, la captura de datos típicamente utiliza uno de dos métodos: basado en registro o basado en disparadores.

    La captura de datos basado en registro implica leer los registros de transacciones de la base de datos para identificar los cambios de datos, que es el método que utilizaremos aquí al utilizar la replicación lógica de Postgres.

    Replicación de Postgres

    Hay dos modos de replicación en Postgres:

    1. Replicación física: cada cambio del primario se transmite a las réplicas a través del WAL (Write Ahead Log). Esta replicación se realiza byte por byte con direcciones de bloque exactas.
    1. Replicación lógica: en la replicación lógica, el suscriptor recibe cada cambio de transacción individual (es decir, declaraciones INSERT, UPDATE o DELETE) en la base de datos.

    El WAL todavía se transmite, pero codifica las operaciones lógicas para que puedan ser decodificadas por el suscriptor sin tener que conocer los detalles internos de Postgres.

    Una de las grandes ventajas de la replicación lógica es que se puede utilizar para replicar sólo tablas o filas específicas, lo que significa que se tiene un control completo sobre lo que se está replicando.

    Para habilitar la replicación lógica, el wal_level debe ser configurado:

    -- determines how much information is written to the wal. 
    -- Each 'level' inherits the level below it; 'logical' is the highest level
    
    ALTER SYSTEM SET wal_level=logical;
    
    -- simultaneously running WAL sender processes
    ALTER SYSTEM SET max_wal_senders='10';
    
    -- simultaneously defined replication slots
    ALTER SYSTEM SET max_replication_slots='10';
    

    Los cambios requieren un reinicio de la instancia de Postgres.

    Después de reiniciar el sistema, el wal_level se puede verificar con:

    SHOW wal_level;
     wal_level 
    -----------
     logical
    (1 row)
    

    Para suscribirse a los cambios se debe crear una publicación . Una publicación es un grupo de tablas en las que nos gustaría recibir cambios de datos.

    Vamos a crear una tabla simple y definir una publicación para ella:

    CREATE TABLE articles (id serial PRIMARY KEY, title text, description text, body text);
    CREATE PUBLICATION articles_pub FOR TABLE articles;
    

    Para indicar a Postgres que retenga segmentos de WAL, debemos crear un slot de replicación.

    El slot de replicación representa un flujo de cambios desde una o más publicaciones y se utiliza para prevenir la pérdida de datos en caso de una falla del servidor, ya que son a prueba de fallos.

    Protocolo de Replicación

    Para tener una idea del protocolo y los mensajes que se envían, podemos usar pg_recvlogical para iniciar un suscriptor de replicación:

    # Start and use the publication defined above
    # output is written to stdout
    pg_recvlogical --start \
      --host='localhost' \
      --port='5432' \
      --username='postgres' \
      --dbname='postgres' \
      --option=publication_names='articles_pub' \
      --option=proto_version=1 \
      --create-slot \
      --if-not-exists \
      --slot=articles_slot \
      --plugin=pgoutput \
      --file=-
    

    Insertar un registro:

    INSERT INTO articles (title, description, body)
        VALUES ('Postgres replication', 'Using logical replication', 'Foo bar baz');
    

    Cada linea en la salida corresponde a un mensaje de replicación recibido a través de suscripción:

    B(egin) - Begin transaction 
    R(elation) - Table, schema, columns and their types
    I(insert) - Data being inserted
    C(ommit) - Commit transaction
    
    ___________________________________
    
    B
    
    Rarticlesdidtitledescriptionbody
    It35tPostgres replicationtUsing logical replicationtFoo bar baz
    C
    

    Si insertamos múltiples registros en una transacción deberíamos tener dos I entre B y C:

    BEGIN;
    INSERT INTO articles (title, description, body) VALUES ('First', 'desc', 'Foo');
    
    INSERT INTO articles (title, description, body) VALUES ('Second', 'desc', 'Bar');
    COMMIT;
    

    Y la salida:

    C
    B
    
    It37tFirsttdesctFoo
    It38tSecondtdesctBar
    CCopied to clipboard!
    

    La información de la relación, es decir, la tabla, no se transmitió porque ya se recibió la relación al insertar el primer registro.

    Postgres solo envía la relación la primera vez que se encuentra durante la sesión. Se espera que el suscriptor almacene en caché una relación previamente enviada.

    Ahora que tenemos una idea de cómo funciona la replicación lógica, ¡implementémosla en Elixir!

    Implementando la conexión de replicación

    Cree un nuevo proyecto de Elixir:

    mix new cdc
    

    Añadiremos las siguientes dependencias a mix.exs:

    defp deps do
      {:postgrex, "~> 0.16.4"},
      # decode/encode replication messages
      {:postgrex_pgoutput, "~> 0.1.0"}
    end
    

    Postgrex admite la replicación a través del proceso Postgrex.ReplicationConnection.

    defmodule CDC.Replication do
      use Postgrex.ReplicationConnection
      require Logger
    
      defstruct [
        :publications,
        :slot,
        :state
      ]
    
      def start_link(opts) do
        conn_opts = [auto_reconnect: true]
        publications = opts[:publications] || raise ArgumentError, message: "`:publications` is required"
        slot = opts[:slot] || raise ArgumentError, message: "`:slot` is required"
    
        Postgrex.ReplicationConnection.start_link(
          __MODULE__,
          {slot, publications},
          conn_opts ++ opts
        )
      end
    
      @impl true
      def init({slot, pubs}) do
        {:ok, %__MODULE__{slot: slot, publications: pubs}}
      end
    
      
      @impl true
      def handle_connect(%__MODULE__{slot: slot} = state) do
        query = "CREATE_REPLICATION_SLOT #{slot} TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT"
    
        Logger.debug("[create slot] query=#{query}")
    
        {:query, query, %{state | state: :create_slot}}
      end
    
      @impl true
      def handle_result([%Postgrex.Result{} | _], %__MODULE__{state: :create_slot, publications: pubs, slot: slot} = state) do
        opts = [proto_version: 1, publication_names: pubs]
    
        query = "START_REPLICATION SLOT #{slot} LOGICAL 0/0 #{escape_options(opts)}"
    
        Logger.debug("[start streaming] query=#{query}")
    
        {:stream, query, [], %{state | state: :streaming}}
      end
    
      @impl true
      def handle_data(msg, state) do
        Logger.debug("Received msg=#{inspect(msg, limit: :infinity, pretty: true)}")
        {:noreply, [], state}
      end
    
      defp escape_options([]),
        do: ""
    
      defp escape_options(opts) do
        parts =
          Enum.map_intersperse(opts, ", ", fn {k, v} -> [Atom.to_string(k), ?\s, escape_string(v)] end)
    
        [?\s, ?(, parts, ?)]
      end
    
      defp escape_string(value) do
        [?', :binary.replace(to_string(value), "'", "''", [:global]), ?']
      end
    end
    

    El código esta disponible en GitHub

    Probemos:

    opts = [
      slot: "articles_slot_elixir",
      publications: ["articles_pub"],
      host: "localhost",
      database: "postgres",
      username: "postgres",
      password: "postgres",
      port: 5432,
    ]
    
    CDC.Replication.start_link(opts)
    

    Cuando iniciamos el proceso, ocurre lo siguiente:

    1. Una vez que estamos conectados a Postgres, se llama al callback handle_connect/1, se crea un slot de replicación lógica temporal.
    2. Se llama a handle_result/2 con el resultado de la consulta en el paso 1. Si el slot se creó correctamente, comenzamos a transmitir desde el slot y entramos en el modo de transmisión. La posición solicitada ‘0/0’ significa que Postgres elige la posición.
    3. Cualquier mensaje de replicación enviado desde Postgres se recibe en el callback handle_data/2.

    Mensajes de replicación

    Hay dos tipos de mensajes que un suscriptor recibe:

    1. primary_keep_alive: un mensaje de comprobación, si reply == 1 se espera que el suscriptor responda al mensaje con un standby_status_update para evitar una desconexión por tiempo de espera.

    El standby_status_update contiene el LSN actual que el suscriptor ha procesado.

    Postgres utiliza este mensaje para determinar qué segmentos de WAL se pueden eliminar de forma segura.

    1. xlog_data: contiene los mensajes de datos para cada paso en una transacción.Dado que no estamos respondiendo a los mensajes primary_keep_alive, el proceso se desconecta y se reinicia.

    Arreglemos esto decodificando los mensajes y comenzando a responder con mensajes standby_status_update.

    
    defmodule CDC.Replication do
      use Postgrex.ReplicationConnection
    
      require Postgrex.PgOutput.Messages
      alias Postgrex.PgOutput.{Messages, Lsn}
    
      require Logger
    
      defstruct [
        :publications,
        :slot,
        :state
      ]
    
      def start_link(opts) do
        conn_opts = [auto_reconnect: true]
        publications = opts[:publications] || raise ArgumentError, message: "`:publications` is required"
        slot = opts[:slot] || raise ArgumentError, message: "`:slot` is required"
    
        Postgrex.ReplicationConnection.start_link(
          __MODULE__,
          {slot, publications},
          conn_opts ++ opts
        )
      end
    
      @impl true
      def init({slot, pubs}) do
        {:ok, %__MODULE__{slot: slot, publications: pubs}}
      end
    
      @impl true
      def handle_connect(%__MODULE__{slot: slot} = state) do
        query = "CREATE_REPLICATION_SLOT #{slot} TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT"
    
        Logger.debug("[create slot] query=#{query}")
    
        {:query, query, %{state | state: :create_slot}}
      end
    
      @impl true
      def handle_result(
            [%Postgrex.Result{} | _],
            %__MODULE__{state: :create_slot, publications: pubs, slot: slot} = state
          ) do
        opts = [proto_version: 1, publication_names: pubs]
    
        query = "START_REPLICATION SLOT #{slot} LOGICAL 0/0 #{escape_options(opts)}"
    
        Logger.debug("[start streaming] query=#{query}")
    
        {:stream, query, [], %{state | state: :streaming}}
      end
    
      @impl true
      def handle_data(msg, state) do
        return_msgs =
          msg
          |> Messages.decode()
          |> handle_msg()
    
        {:noreply, return_msgs, state}
      end
    
      #
      defp handle_msg(Messages.msg_primary_keep_alive(server_wal: lsn, reply: 1)) do
        Logger.debug("msg_primary_keep_alive message reply=true")
        <<lsn::64>> = Lsn.encode(lsn)
    
        [standby_status_update(lsn)]
      end
    
      defp handle_msg(Messages.msg_primary_keep_alive(reply: 0)), do: []
    
      defp handle_msg(Messages.msg_xlog_data(data: data)) do
        Logger.debug("xlog_data message: #{inspect(data, pretty: true)}")
        []
      end
    
      defp standby_status_update(lsn) do
        [
          wal_recv: lsn + 1,
          wal_flush: lsn + 1,
          wal_apply: lsn + 1,
          system_clock: Messages.now(),
          reply: 0
        ]
        |> Messages.msg_standby_status_update()
        |> Messages.encode()
      end
    
      
    defp escape_options([]),
        do: ""
    
      defp escape_options(opts) do
        parts =
          Enum.map_intersperse(opts, ", ", fn {k, v} -> [Atom.to_string(k), ?\s, escape_string(v)] end)
    
        [?\s, ?(, parts, ?)]
      end
    
      defp escape_string(value) do
        [?', :binary.replace(to_string(value), "'", "''", [:global]), ?']
      end
    end
    

    handle_data/2 decodifica el mensaje y lo pasa a handle_msg/1. Si es un primary_keep_alive , respondemos con un standby_status_update.

    El LSN denota una posición de byte en el WAL.

    El suscriptor responde con el LSN que ha manejado actualmente, como no estamos haciendo seguimiento de los mensajes que recibimos, simplemente confirmamos con el LSN enviado desde el servidor.

    A continuación, manejaremos los mensajes xlog_data, la idea aquí es que capturaremos cada operación en una estructura de transacción.

    Capturando transacciones

    El módulo CDC.Protocol manejará los mensajes xlog_data y rastreará el estado de la transacción

    defmodule CDC.Protocol do
      import Postgrex.PgOutput.Messages
      require Logger
    
      alias CDC.Tx
      alias Postgrex.PgOutput.Lsn
    
      
    @type t :: %__MODULE__{
              tx: Tx.t(),
              relations: map()
            }
    
      defstruct [
        :tx,
        relations: %{}
      ]
    
      @spec new() :: t()
      def new do
        %__MODULE__{}
      end
    
      def handle_message(msg, state) when is_binary(msg) do
        msg
        |> decode()
        |> handle_message(state)
      end
    
      def handle_message(msg_primary_keep_alive(reply: 0), state), do: {[], nil, state}
      def handle_message(msg_primary_keep_alive(server_wal: lsn, reply: 1), state) do
        Logger.debug("msg_primary_keep_alive message reply=true")
        <<lsn::64>> = Lsn.encode(lsn)
    
        {[standby_status_update(lsn)], nil, state}
      end
    
      def handle_message(msg, %__MODULE__{tx: nil, relations: relations} = state) do
        tx =
          [relations: relations, decode: true]
          |> Tx.new()
          |> Tx.build(msg)
    
        {[], nil, %{state | tx: tx}}
      end
    
      def handle_message(msg, %__MODULE__{tx: tx} = state) do
        case Tx.build(tx, msg) do
          %Tx{state: :commit, relations: relations} ->
            tx = Tx.finalize(tx)
            relations = Map.merge(state.relations, relations)
            {[], tx, %{state | tx: nil, relations: relations}}
    
          tx ->
            {[], nil, %{state | tx: tx}}
        end
      end
    
      defp standby_status_update(lsn) do
        [
          wal_recv: lsn + 1,
          wal_flush: lsn + 1,
          wal_apply: lsn + 1,
          system_clock: now(),
          reply: 0
        ]
        |> msg_standby_status_update()
        |> encode()
      end
    end
    

    CDC.Tx maneja mensajes recibidos dentro de la transacción, begin, relation, insert/update/delete y commit.

    defmodule CDC.Tx do
      import Postgrex.PgOutput.Messages
      alias Postgrex.PgOutput.Lsn
    
      alias __MODULE__.Operation
    
      @type t :: %__MODULE__{
              operations: [Operation.t()],
              relations: map(),
              timestamp: term(),
              xid: pos_integer(),
              state: :begin | :commit,
              lsn: Lsn.t(),
              end_lsn: Lsn.t()
            }
    
      defstruct [
        :timestamp,
        :xid,
        :lsn,
        :end_lsn,
        relations: %{},
        operations: [],
        state: :begin,
        decode: true
      ]
    
      def new(opts \\ []) do
        struct(__MODULE__, opts)
      end
    
      def finalize(%__MODULE__{state: :commit, operations: ops} = tx) do
        %{tx | operations: Enum.reverse(ops)}
      end
    
      def finalize(%__MODULE__{} = tx), do: tx
    
      @spec build(t(), tuple()) :: t()
      def build(tx, msg_xlog_data(data: data)) do
        build(tx, data)
      end
    
      def build(tx, msg_begin(lsn: lsn, timestamp: ts, xid: xid)) do
        %{tx | lsn: lsn, timestamp: ts, xid: xid, state: :begin}
      end
    
      def build(%__MODULE__{state: :begin, relations: relations} = tx, msg_relation(id: id) = rel) do
        %{tx | relations: Map.put(relations, id, rel)}
      end
    
      def build(%__MODULE__{state: :begin, lsn: tx_lsn} = tx, msg_commit(lsn: lsn, end_lsn: end_lsn))
          when tx_lsn == lsn do
        %{tx | state: :commit, end_lsn: end_lsn}
      end
    
      def build(%__MODULE__{state: :begin} = builder, msg_insert(relation_id: id) = msg),
        do: build_op(builder, id, msg)
    
      def build(%__MODULE__{state: :begin} = builder, msg_update(relation_id: id) = msg),
        do: build_op(builder, id, msg)
    
      def build(%__MODULE__{state: :begin} = builder, msg_delete(relation_id: id) = msg),
        do: build_op(builder, id, msg)
    
      # skip unknown messages
      def build(%__MODULE__{} = tx, _msg), do: tx
    
      defp build_op(%__MODULE__{state: :begin, relations: rels, decode: decode} = tx, id, msg) do
        rel = Map.fetch!(rels, id)
        op = Operation.from_msg(msg, rel, decode)
    
        %{tx | operations: [op | tx.operations]}
      end
    end
    

    CDC.Tx.Operation maneja los mensajes INSERT/UPDATE/DELETE y decodifica los datos combinándolos con la relación

    defmodule CDC.Tx.Operation do
      @moduledoc "Describes a change (INSERT, UPDATE, DELETE) within a transaction."
    
      import Postgrex.PgOutput.Messages
      alias Postgrex.PgOutput.Type, as: PgType
    
      @type t :: %__MODULE__{}
      defstruct [
        :type,
        :schema,
        :namespace,
        :table,
        :record,
        :old_record,
        :timestamp
      ]
    
      @spec from_msg(tuple(), tuple(), decode :: boolean()) :: t()
      def from_msg(
            msg_insert(data: data),
            msg_relation(columns: columns, namespace: ns, name: name),
            decode?
          ) do
        %__MODULE__{
          type: :insert,
          namespace: ns,
          schema: into_schema(columns),
          table: name,
          record: cast(data, columns, decode?),
          old_record: %{}
        }
      end
    
      def from_msg(
            msg_update(change_data: data, old_data: old_data),
            msg_relation(columns: columns, namespace: ns, name: name),
            decode?
          ) do
        %__MODULE__{
          type: :update,
          namespace: ns,
          table: name,
          schema: into_schema(columns),
          record: cast(data, columns, decode?),
          old_record: cast(columns, old_data, decode?)
        }
      end
    
      def from_msg(
            msg_delete(old_data: data),
            msg_relation(columns: columns, namespace: ns, name: name),
            decode?
          ) do
        %__MODULE__{
          type: :delete,
          namespace: ns,
          schema: into_schema(columns),
          table: name,
          record: %{},
          old_record: cast(data, columns, decode?)
        }
      end
    
      defp into_schema(columns) do
        for c <- columns do
          c
          |> column()
          |> Enum.into(%{})
        end
      end
    
      defp cast(data, columns, decode?) do
        Enum.zip_reduce([data, columns], %{}, fn [text, typeinfo], acc ->
          key = column(typeinfo, :name)
    
          value =
            if decode? do
              t =
                typeinfo
                |> column(:type)
                |> PgType.type_info()
    
              PgType.decode(text, t)
            else
              text
            end
    
          Map.put(acc, key, value)
        end)
      end
    end
    

    Como antes, el mensaje primary_keep_alive con reply == 1 envía un standby_status_update. Cuando recibimos un mensaje xlog_data, creamos un nuevo %Tx{} que usamos para “construir” la transacción hasta que recibimos un msg_commit que marca el final de la transacción.

    Cualquier mensaje de inserción, actualización o eliminación crea una CDC.Tx.Operation en la transacción, cada operación contiene un relation_id que se utiliza para buscar la relación desde tx.relations.

    La operación junto con la relación nos permite decodificar los datos. La información de columna y tipo se recupera de la relación y se utiliza para decodificar los valores en términos de Elixir.

    Una vez que estamos en un estado de commit, fusionamos Tx.relations con Protocol.relations, ya que un mensaje de relación sólo se transmitirá la primera vez que se encuentre una tabla durante la sesión de conexión, Protocol.relations contiene todos los msg_relation que se nos han enviado durante la sesión.

    El módulo CDC.Replication ahora se ve así:

    defmodule CDC.Replication do
      use Postgrex.ReplicationConnection
    
      alias CDC.Protocol
    
      require Logger
    
      defstruct [
        :publications,
        :protocol,
        :slot,
        :state
      ]
    
      def start_link(opts) do
        conn_opts = [auto_reconnect: true]
        publications = opts[:publications] || raise ArgumentError, message: "`:publications` is required"
        slot = opts[:slot] || raise ArgumentError, message: "`:slot` is required"
    
        Postgrex.ReplicationConnection.start_link(
          __MODULE__,
          {slot, publications},
          conn_opts ++ opts
        )
      end
    
      @impl true
      def init({slot, pubs}) do
        {:ok,
         %__MODULE__{
           slot: slot,
           publications: pubs,
           protocol: Protocol.new()
         }}
      end
    
      @impl true
      def handle_connect(%__MODULE__{slot: slot} = state) do
        query = "CREATE_REPLICATION_SLOT #{slot} TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT"
    
        Logger.debug("[create slot] query=#{query}")
    
        {:query, query, %{state | state: :create_slot}}
      end
    
      @impl true
      def handle_result(
            [%Postgrex.Result{} | _],
            %__MODULE__{state: :create_slot, publications: pubs, slot: slot} = state
          ) do
        opts = [proto_version: 1, publication_names: pubs]
    
        query = "START_REPLICATION SLOT #{slot} LOGICAL 0/0 #{escape_options(opts)}"
    
        Logger.debug("[start streaming] query=#{query}")
    
        {:stream, query, [], %{state | state: :streaming}}
      end
    
      @impl true
      def handle_data(msg, state) do
        {return_msgs, tx, protocol} = Protocol.handle_message(msg, state.protocol)
    
        if not is_nil(tx) do
          Logger.debug("Tx: #{inspect(tx, pretty: true)}")
        end
    
        {:noreply, return_msgs, %{state | protocol: protocol}}
      end
    
      
    defp escape_options([]),
        do: ""
    
      defp escape_options(opts) do
        parts =
          Enum.map_intersperse(opts, ", ", fn {k, v} -> [Atom.to_string(k), ?\s, escape_string(v)] end)
    
        [?\s, ?(, parts, ?)]
      end
    
      defp escape_string(value) do
        [?', :binary.replace(to_string(value), "'", "''", [:global]), ?']
      end
    end
    

    handle_data/2 llama a Protocol.handle_message/1 que devuelve una tupla con tres elementos {messages_to_send :: [binary()], complete_transaction :: CDC.Tx.t() | nil, CDC.Protocolo.t()}

    Por ahora solo inspeccionamos la transacción cuando se emite desde Protocol.handle_message/3, probémoslo:

    Interactive Elixir (1.14.0) - press Ctrl+C to exit (type h() ENTER for help)
    opts = [
      slot: "articles_slot_elixir",
      publications: ["articles_pub"],
      host: "localhost",
      database: "postgres",
      username: "postgres",
      password: "postgres",
      port: 5432,
    ]
    
    {:ok, _} = CDC.Replication.start_link(opts)
    {:ok, pid} = Postgrex.start_link(opts)
    
    insert_query = """
    INSERT INTO articles (title, description, body) 
    VALUES ('Postgres replication', 'Using logical replication', 'with Elixir!')
    """
    
    _ = Postgrex.query!(pid, insert_query, [])
      
    14:03:48.020 [debug] Tx: %CDC.Tx{
      timestamp: ~U[2022-10-31 13:03:48Z],
      xid: 494,
      lsn: {0, 22981920},
      end_lsn: nil,
      relations: %{
        16386 => {:msg_relation, 16386, "public", "articles", :default,
         [
           {:column, [:key], "id", :int4, -1},
           {:column, [], "title", :text, -1},
           {:column, [], "description", :text, -1},
           {:column, [], "body", :text, -1}
         ]}
      },
      operations: [
        %CDC.Tx.Operation{
          type: :insert,
          schema: [
            %{flags: [:key], modifier: -1, name: "id", type: :int4},
            %{flags: [], modifier: -1, name: "title", type: :text},
            %{flags: [], modifier: -1, name: "description", type: :text},
            %{flags: [], modifier: -1, name: "body", type: :text}
          ],
          namespace: "public",
          table: "articles",
          record: %{
            "body" => "with Elixir!",
            "description" => "Using logical replication",
            "id" => 6,
            "title" => "Postgres replication"
          },
          old_record: %{},
          timestamp: nil
        }
      ],
      state: :begin,
      decode: true
    }
    

    Cada cambio en la transacción se almacena en Tx.operations, operation.record es la fila decodificada como un mapa.

    Finalmente, implementemos una forma de suscribirnos a los cambios de CDC.Replication:

    defmodule CDC.Replication do
      use Postgrex.ReplicationConnection
    
      alias CDC.Protocol
    
      require Logger
    
      defstruct [
        :publications,
        :protocol,
        :slot,
        :state,
        subscribers: %{}
      ]
    
      def start_link(opts) do
        conn_opts = [auto_reconnect: true]
        publications = opts[:publications] || raise ArgumentError, message: "`:publications` is required"
        slot = opts[:slot] || raise ArgumentError, message: "`:slot` is required"
    
        Postgrex.ReplicationConnection.start_link(
          __MODULE__,
          {slot, publications},
          conn_opts ++ opts
        )
      end
    
      def subscribe(pid, opts \\ []) do
        Postgrex.ReplicationConnection.call(pid, :subscribe, Keyword.get(opts, :timeout, 5_000))
      end
    
      def unsubscribe(pid, ref, opts \\ []) do
        Postgrex.ReplicationConnection.call(
          pid,
          {:unsubscribe, ref},
          Keyword.get(opts, :timeout, 5_000)
        )
      end
    
      @impl true
      def init({slot, pubs}) do
        {:ok,
         %__MODULE__{
           slot: slot,
           publications: pubs,
           protocol: Protocol.new()
         }}
      end
    
      @impl true
      def handle_connect(%__MODULE__{slot: slot} = state) do
        query = "CREATE_REPLICATION_SLOT #{slot} TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT"
    
        Logger.debug("[create slot] query=#{query}")
    
        {:query, query, %{state | state: :create_slot}}
      end
    
      @impl true
      def handle_result(
            [%Postgrex.Result{} | _],
            %__MODULE__{state: :create_slot, publications: pubs, slot: slot} = state
          ) do
        opts = [proto_version: 1, publication_names: pubs]
    
        query = "START_REPLICATION SLOT #{slot} LOGICAL 0/0 #{escape_options(opts)}"
    
        Logger.debug("[start streaming] query=#{query}")
    
        {:stream, query, [], %{state | state: :streaming}}
      end
    
      @impl true
      def handle_data(msg, state) do
        {return_msgs, tx, protocol} = Protocol.handle_message(msg, state.protocol)
    
        if not is_nil(tx) do
          notify(tx, state.subscribers)
        end
    
        {:noreply, return_msgs, %{state | protocol: protocol}}
      end
    
      # Replies must be sent using `reply/2`
      # https://hexdocs.pm/postgrex/Postgrex.ReplicationConnection.html#reply/2
      @impl true
      def handle_call(:subscribe, {pid, _} = from, state) do
        ref = Process.monitor(pid)
    
        state = put_in(state.subscribers[ref], pid)
    
        Postgrex.ReplicationConnection.reply(from, {:ok, ref})
    
        {:noreply, state}
      end
    
      def handle_call({:unsubscribe, ref}, from, state) do
        {reply, new_state} =
          case state.subscribers do
            %{^ref => _pid} ->
              Process.demonitor(ref, [:flush])
    
              {_, state} = pop_in(state.subscribers[ref])
              {:ok, state}
    
            _ ->
              {:error, state}
          end
    
        from && Postgrex.ReplicationConnection.reply(from, reply)
    
        {:noreply, new_state}
      end
    
      @impl true
      def handle_info({:DOWN, ref, :process, _, _}, state) do
        handle_call({:unsubscribe, ref}, nil, state)
      end
    
      defp notify(tx, subscribers) do
        for {ref, pid} <- subscribers do
          send(pid, {:notification, self(), ref, tx})
        end
    
        :ok
      end
    
      defp escape_options([]),
        do: ""
    
      defp escape_options(opts) do
        parts =
          Enum.map_intersperse(opts, ", ", fn {k, v} -> [Atom.to_string(k), ?\s, escape_string(v)] end)
    
        [?\s, ?(, parts, ?)]
      end
    
      defp escape_string(value) do
        [?', :binary.replace(to_string(value), "'", "''", [:global]), ?']
      end
    end
    
    

    Y lo podemos usar así:

    opts = [
      slot: "articles_slot",
      publications: ["articles_pub"],
      host: "localhost",
      database: "postgres",
      username: "postgres",
      password: "postgres",
      port: 5432,
    ]
    
    {:ok, pid} = CDC.Replication.start_link(opts)
    {:ok, pg_pid} = Postgrex.start_link(opts)
    {:ok, ref} = CDC.Replication.subscribe(pid)
    
    insert_query = """
    INSERT INTO articles (title, description, body) 
    VALUES ('Postgres replication', 'Using logical replication', 'with Elixir!')
    """
    
    _ = Postgrex.query!(pg_pid, insert_query, [])
    flush()
    
    {:notification, #PID<0.266.0>, #Reference<0.2499916608.3416784901.94813>,
     %CDC.Tx{
       timestamp: ~U[2022-10-31 13:26:35Z],
       xid: 495,
       lsn: {0, 22983536},
       end_lsn: nil,
       relations: %{
         16386 => {:msg_relation, 16386, "public", "articles", :default,
          [
            {:column, [:key], "id", :int4, -1},
            {:column, [], "title", :text, -1},
            {:column, [], "description", :text, -1},
            {:column, [], "body", :text, -1}
          ]}
       },
       operations: [
         %CDC.Tx.Operation{
           type: :insert,
           schema: [
             %{flags: [:key], modifier: -1, name: "id", type: :int4},
             %{flags: [], modifier: -1, name: "title", type: :text},
             %{flags: [], modifier: -1, name: "description", type: :text},
             %{flags: [], modifier: -1, name: "body", type: :text}
           ],
           namespace: "public",
           table: "articles",
           record: %{
             "body" => "with Elixir!",
             "description" => "Using logical replication",
             "id" => 7,
             "title" => "Postgres replication"
           },
           old_record: %{},
           timestamp: nil
         }
       ],
       state: :begin,
       decode: true
     }}
    

    Conclusión

    Si está buscando una manera de capturar cambios de su base de datos con cambios mínimos en su configuración existente, definitivamente vale la pena considerar Cambiar la captura de datos. Con Elixir y postgrex hemos implementado un mini Debezium en ~400 LOC. La fuente completa está disponible aquí .

    Si necesita ayuda con la implementación de Elixir, nuestro equipo de expertos líder en el mundo siempre está aquí para ayudarlo. Contáctenos hoy para saber cómo podemos ayudarlo.

    The post Captura de datos con Postgres y Elixir appeared first on Erlang Solutions .

    • chevron_right

      Ignite Realtime Blog: Spark 3.0.2 Released

      news.movim.eu / PlanetJabber · Friday, 31 March, 2023 - 18:16 · 1 minute

    The Ignite Realtime community is happy to announce the availability of Spark version 3.0.2

    The release contains bug fixes and updates two plugins Translator and Roar.

    Many Spark translations are incomplete. Please help us translate Spark

    Full list of changes can be found in the changelog .

    We encourage users and developers to get involved with Spark project by providing feedback in the forums or submitting pull requests on our GitHub page.

    You can download Spark from the Downloads page. Below are the sha256 checksums:

    db7febafd05a064ffed4800066d11d0a2d27288aa3c0b174ed21a20672ab4669 *spark_3_0_2-64bit.exe
    647712b43942f7dc399901c9078ea1de7fcca14de3b5898bc3bdeee7751cf1b3 *spark_3_0_2-64bit.msi
    c1415417081095656c5fd7eb42e09ac6168f550af3c5254eb35506a404283a3c *spark_3_0_2-arm.exe
    78994eaa31d6c074bc72b9ac7a1f38f63488f9272c1a8870524f986dda618b6b *spark_3_0_2-with-jre.dmg
    ef3ba8eef5b88edc5e4ce9e13e9fa41ef2fad136cc6b518c52da79051c2a7c39 *spark_3_0_2-with-jre.exe
    d18dd7613333647f3fae6728731df7e3ef957103a84686ce1bb7befb9d32285f *spark_3_0_2-with-jre.msi
    39b03b4f4363bc053d2c5ed0c7452f97ced38a1c307ab77ce7f8b0e36459f498 *spark_3_0_2.deb
    5bc3bf000b4fe8a49d424ea53ccd6b62dae9291e987394f08c62ff0d69f0aec9 *spark_3_0_2.dmg
    e511c10db5f72e8b311f79dc3ac810040d44c5488f606cb860d7250d1dcf3709 *spark_3_0_2.exe
    f54d3990dd0ca76a1ca447af0bef9b4244861c7da9f6ab38a755c9cc578344c8 *spark_3_0_2.msi
    fff7fa157fe187b5307ef2383c87b8193a0416ffe41ffcb2ea0b0e6672a917f9 *spark_3_0_2.rpm
    a5cf06ccbe82dc308f2d2493bc0d89f552729fb993af392003263f0cd9caac16 *spark_3_0_2.sh
    03eae1e4e88fdc303752f3b5c0ef8bb00653cfdf28ee964699bba892e45110e4 *spark_3_0_2.tar.gz
    

    For other release announcements and news follow us on Twitter and Mastodon .

    1 post - 1 participant

    Read full topic

    • chevron_right

      Erlang Solutions: 5 Key Tech Priorities for Fintech Leaders in 2023

      news.movim.eu / PlanetJabber · Thursday, 30 March, 2023 - 10:09 · 6 minutes

    The fintech industry is a major disruptor. Each year, it impacts how consumers interact with financial companies and brings new and innovative means to meet ever-growing customer expectations and occupy market space.

    As a business owner or executive in this space, you have no choice but to stay on top of your game to increase efficiency.

    In simpler terms, if your business doesn’t scale, it could fail.

    That might sound a tad extreme, but you’re dealing in a market that is set to be worth $698.48 billion by 2030. Understanding the trends and focuses of the time helps you know where you’re headed in order to remain competitive.

    So let’s talk about these trends that drive the market.

    Embed finance

    The embedded finance market is expected to see massive growth this year. The market has been growing at a rapid pace since 2020. Fintech companies have been steadily outpacing traditional banking services when it comes to gaining the trust of consumers.

    According to our latest report , searches for the term have increased by a staggering 488% in the last five years.

    Embedded finance solutions can make up 50% of total banking revenue streams in the near future. There is also a significant market for embedded financial products in the areas of deposit accounts, payments, insurance, and lending.

    Source: Mckinsey analysis

    Arguably the greatest benefit to embedding finance is that under the direction of inclusive fintech startups, it has the potential to empower potential customers who were previously excluded from the conventional financial industry. Similarly, emerging markets can provide a less stifling environment with lower prices and a larger customer base, which would further encourage innovation.

    We’re also likely to see traditional financial services and fintech firms, such as banks and payment processors, collaborating more closely to adopt embedded finance. This could result in payment fintechs providing more inclusive services tailored to various business models, leasing more potential while banks provide fundamental infrastructure.

    Embedded finance solutions will place much greater emphasis on technological advantages and operational capability to address your business’s current issues. Think risk and compliance management. Innovations like distributed computing and artificial intelligence (AI) will have multiple effects across all businesses, strengthening their ability to embrace embedded finance.

    AI

    Financial institutions are continuing to adopt AI and machine learning, the industry is set to see a projected savings of up to 22% by the year 2030 .

    How is this possible? AI-driven chatbots and assistants can make the customer experience very efficient. They can answer customer questions, monitor spending and tailor product recommendations based directly on customer preferences.

    While chatbots do not quite replicate the human experience, their growth across the market this year means we can expect more of them in 2023.

    Source: Adobe

    Think about your price comparison sites when looking for insurance, or travel rates. These services have been made possible through the application of Natural Language Processing (NLP), which allows the ability to make payments and offer personalised service at any given time.

    A key element of AI technology is its ability to predict human interactions, combining the best of two words, intelligence and behavioural finance. What is Behavioural Finance? It focuses on human psychology to influence economic markets which in turn, impacts market outcomes. It is considered the future of fintech, as supported by the development of data analytics and large amounts of consumer data. This data, combined with AI algorithms, creates the ability to provide such personalised services.

    The industry is starting to recognize the goldmine that is AI. It doesn’t just drive efficiency but also makes allows businesses to be smarter in their approach to interacting with customers in the market.

    Blockchain

    Blockchain is one of the most exciting trends in Fintech today. It addresses the problem of unsecured, expensive fund transfer processes, with high-speed transactions at a much lower cost.

    It works as a digital ledger that verifies, records, and facilitates various types of transactions. Its major selling point is security and autonomy for customers, as businesses and individuals can safely transfer digital assets without relying on a central authority or financial institutions.

    But this isn’t just the only thing that appeals to users.

    Blockchain is used in various fintech applications, such as:

    • Decentralised finance (DeFi): Creates decentralised financial services- for example borrowing, health insurance, lending services etc
    • Digital assets: Provides a secure, transparent method of storing and trading digital assets- got example, NFTs and cryptocurrencies.
    • Cross-border payment services: Enable secure and fast cross-border payments
    • Supply chain management: Provides transparent and secure methods of tracking goods and assets within supply chain networks.
    • Identity management- Creates decentralised identity systems, offering increased security and privacy.

    As blockchain eliminates the need for traditional banking, it now makes it easier for customers to utilise these financial offerings. This is particularly significant for third-world and developing countries, where access to traditional banks can sometimes be challenging.

    Customer Experience

    We’ve spoken a lot about customer experience expectations already, but it is integral to the future of fintech services. After all, the focus will always be on the people your business serves.

    Banks that consistently optimise the customer experience are set to grow 3.2x faster than their competitors.

    It’s up to businesses to provide seamless, user-friendly experiences to attract and retain their customers. This means focusing investment on user-centred design, more personalised services and omnichannel support.

    Put yourselves in your customer’s shoes to understand their pain points, and use that data to create customised solutions that far exceed expectations.

    Sustainability and ESG

    Any forward-thinking businesses will have Environmental, Social and Corporate Governance (ESG) and sustainability at the heart of their plans this year.

    Customers, investors and employees alike are keen to see businesses contribute to making society more sustainable. And with a net-zero goal to be met by 2050 (in the UK), organisations across the globe are under much greater scrutiny to implement supporting policies.

    According to FIS Global , “ESG is top of mind for financial services firms globally, with 60% of executives saying they are developing new ESG products and services.”

    So if your business isn’t environmentally friendly, it might have an impact on your customer base. Brands that care about the planet and show strategic planning are the ones set to thrive. If you don’t care, expect your customers to find a business that does.

    This increased sustainability interest has led to a lot of financial institutions offering a diverse portfolio of sustainable options, such as loans backed by environmental enterprises, such as investment into wind or solar farms. ESG bonds are also rising in popularity.

    A trend we’ll see this year will be sustainable (green) financing. This is another way that financial regulations and products are orchestrated to meet environmentally sustainable outcomes.

    Green Financing will be a key Fintech trend that will aid the private sector in doing positive work for the environment. It will also encourage private-public alliances on financial mechanisms like green bonds.

    Looking ahead

    With the ongoing development of the Fintech industry, technological innovation must work in tandem with the economy’s call for greater financial inclusion.

    As firms across the globe embrace the potential of tech to champion the future, consider these trends and how (or if) your business is incorporating them.

    Want a more in-depth look into fintech themes and how they should inform your tech strategy and decision-making? You can download our latest whitepaper to hear from the experts.

    Our team are also on hand if you’d like a chat about your fintech project or any projects you might have. Don’t hesitate to contact us .

    The post 5 Key Tech Priorities for Fintech Leaders in 2023 appeared first on Erlang Solutions .

    • wifi_tethering open_in_new

      This post is public

      www.erlang-solutions.com /blog/5-key-tech-priorities-for-fintech-leaders-in-2023/

    • chevron_right

      Erlang Solutions: Cómo depurar tu RabbitMQ

      news.movim.eu / PlanetJabber · Wednesday, 29 March, 2023 - 11:08 · 14 minutes

    Descubre las herramientas y métodos adecuados para la depuración de RabbitMQ.

    Lo que aprenderás en este blog.

    Nuestros clientes de consultoría de RabbitMQ provienen de una amplia gama de industrias. Como resultado, hemos visto casi todos los comportamientos inesperados que puede presentar. RabbitMQ es un software complejo que emplea concurrencia y cómputo distribuido (a través de Erlang), por lo que depurarlo no siempre es sencillo. Para llegar a la causa raíz de un comportamiento inesperado (y no deseado), necesitas las herramientas adecuadas y la metodología adecuada. En este artículo, demostraremos ambas para ayudarte a aprender la técnica de depuración en RabbitMQ.

    El problema de depurar RabbitMQ.

    La inspiración para este blog proviene de un ejemplo real. Uno de nuestros clientes tenía la API HTTP de administración de RabbitMQ proporcionando información crucial a su sistema. El sistema dependía mucho de la API, específicamente del endpoint /api/queues  porque el sistema necesitaba saber el número de mensajes listos en cada cola en un clúster de RabbitMQ. El problema era que a veces una solicitud HTTP al endpoint duraba hasta decenas de segundos (en el peor de los casos, ni siquiera podían obtener una respuesta de la API).

    Entonces, ¿qué causó que algunas solicitudes tomaran tanto tiempo? Para responder a esa pregunta, intentamos reproducir el problema a través de pruebas de carga.

    Ejecutando pruebas de carga.

    Utilizamos una plataforma que creamos para MongooseIM para ejecutar nuestras Pruebas de Carga Continuas . Aquí están algunos de los aspectos más importantes de la plataforma:

    1. todos los servicios involucrados en una prueba de carga se ejecutan dentro de contenedores de docker
    2. la carga es generada por Amoc ; es una herramienta de código abierto escrita en Erlang para generar cargas masivamente paralelas de cualquier tipo (AMQP en nuestro caso)
    3. se recopilan métricas del sistema en prueba y del sitio de Amoc para un análisis posterior.

    El diagrama a continuación representa una arquitectura lógica de una prueba de carga de ejemplo con RabbitMQ:

    En el diagrama, el lado izquierdo muestra un clúster de nodos de Amoc que emulan clientes AMQP que, a su vez, generan la carga contra RabbitMQ. Por otro lado, podemos ver un clúster de RabbitMQ que sirve a los clientes AMQP. Todas las métricas de los servicios de Amoc y RabbitMQ se recopilan y almacenan en una base de datos InfluxDB.

    Consultas lentas de Management HTTP API

    Intentamos reproducir las consultas lentas a Management HTTP API en nuestras pruebas de carga. El escenario de prueba fue bastante sencillo. Un grupo de editores publicaba mensajes en el intercambio predeterminado. Los mensajes de cada editor se dirigían a una cola dedicada (cada editor tenía una cola dedicada). También había consumidores conectados a cada cola. La replicación de cola estaba habilitada.

    Para valores concretos, consulte la tabla a continuación:

    Esa configuración estresó los servidores Rabbit en nuestra infraestructura. Como se ve en los gráficos a continuación:

    Cada nodo de RabbitMQ consumió alrededor de 6 (de 7) núcleos de CPU y aproximadamente 1,4 GB de RAM, excepto rabbitmq-1 que consumió significativamente más que los otros. Eso se debió probablemente a que tuvo que atender más solicitudes de Management HTTP API que los otros dos nodos.

    Durante la prueba de carga, se consultó el endpoint /api/queues cada dos segundos para obtener la lista de todas las colas junto con los valores correspondientes de messages_ready . Una consulta se veía así:

    http://rabbitmq-1:15672/api/queues?columns=name,messages_ready

    Aquí están los resultados de la prueba:

    La figura anterior muestra el tiempo de consulta durante una prueba de carga. Está claro que las cosas son muy lentas. La mediana es de 1,5 segundos mientras que los percentiles 95, 99, 999 y máx. llegan a 20 segundos.

    Debugging

    Una vez confirmado el problema y puede reproducirse, estamos listos para comenzar a depurar. La primera idea fue encontrar la función Erlang que se llama cuando llega una solicitud a la API de administración HTTP de RabbitMQ y determinar dónde esa función pasa su tiempo de ejecución. Si pudiéramos hacer esto, nos permitiría localizar el código más costoso detrás de la API.

    Encontrar la función de entrada

    Para encontrar la función que estábamos buscando, tomamos los siguientes pasos:

    1. buscamos en el complemento de administración de RabbitMQ para encontrar la asignación adecuada de “ruta HTTP a función”,
    2. usamos la función de rastreo de Erlang para verificar si se llama a una función encontrada cuando llega una solicitud.

    El complemento de administración utiliza cowboy (un servidor HTTP de Erlang) debajo para servir las solicitudes de API. Cada punto final de HTTP requiere un módulo de devolución de llamada de cowboy, por lo que encontramos fácilmente la función rabbit_mgmt_wm_queues:to_json/2 que parecía manejar las solicitudes que llegaban a /api/queues. Confirmamos eso con el rastreo (usando una biblioteca de recuperación que se envía con RabbitMQ por defecto).

    root@rmq-test-rabbitmq-1:/rabbitmq_server-v3.7.9# erl -remsh rabbit@rmq-test-rabbitmq-1 -sname test2 -setcookie rabbit  
    Erlang/OTP 21 [erts-10.1] [source] [64-bit] [smp:22:7] [ds:22:7:10] [async-threads:1]  
    
    Eshell V10.1  (abort with ^G)  
    (rabbit@rmq-test-rabbitmq-1)1> recon_trace:calls({rabbit_mgmt_wm_queues, to_json, 2}, 1).  
    1  
    
    11:0:48.464423 <0.1294.15> rabbit_mgmt_wm_queues:to_json(#{bindings => #{},body_length => 0,cert => undefined,charset => undefined,  
      has_body => false,  
      headers =>  
          #{<<"accept">> => <<"*/*">>,  
            <<"authorization">> => <<"Basic Z3Vlc3Q6Z3Vlc3Q=">>,  
            <<"host">> => <<"10.100.10.140:53553">>,  
            <<"user-agent">> => <<"curl/7.54.0">>},  
      host => <<"10.100.10.140">>,host_info => undefined,  
      media_type => {<<"application">>,<<"json">>,[]},  
      method => <<"GET">>,path => <<"/api/queues">>,path_info => undefined,  
      peer => {{10,100,10,4},54136},  
      pid => <0.1293.15>,port => 53553,qs => <<"columns=name,messages_ready">>,  
      ref => rabbit_web_dispatch_sup_15672,  
      resp_headers =>  
          #{<<"content-security-policy">> => <<"default-src 'self'">>,  
            <<"content-type">> => [<<"application">>,<<"/">>,<<"json">>,<<>>],  
            <<"vary">> =>  
                [<<"accept">>,  
                 [<<", ">>,<<"accept-encoding">>],  
                 [<<", ">>,<<"origin">>]]},  
      scheme => <<"http">>,  
      sock => {{172,17,0,4},15672},  
      streamid => 1,version => 'HTTP/1.1'}, {context,{user,<<"guest">>,  
                   [administrator],  
                   [{rabbit_auth_backend_internal,none}]},  
             <<"guest">>,undefined})  
    Recon tracer rate limit tripped. 
    

    El fragmento anterior muestra que primero habilitamos el seguimiento para rabbit_mgmt_wm_queues:to_json/2 , luego enviamos manualmente una solicitud a la API de administración (usando curl; no visible en el fragmento) y que generó el evento de seguimiento. Así es como encontramos nuestro punto de entrada para un análisis más profundo.

    Usando flame graphs

    Una vez que hemos encontrado una función que sirve las solicitudes, ahora podemos verificar cómo esa función pasa su tiempo de ejecución. La técnica ideal para hacer esto es Flame Graphs. Una de sus definiciones establece que:

    Los gráficos de llamas son una visualización del software perfilado, lo que permite identificar rápidamente y con precisión los caminos de código más frecuentes.

    En nuestro caso, podríamos usar gráficos de llamas para visualizar la pila de llamadas de la función o, en otras palabras, qué funciones se llaman dentro de una función rastreada y cuánto tiempo tarda (en relación con el tiempo de ejecución de la función rastreada) para que estas funciones se ejecuten. Esta visualización ayuda a identificar rápidamente las funciones sospechosas en el código.

    Para Erlang, existe una biblioteca llamada eflame que tiene herramientas tanto para: recopilar trazas del sistema Erlang como para construir un gráfico de llamas a partir de los datos. ¿Pero cómo inyectamos esa biblioteca en Rabbit para nuestra prueba de carga?

    Construyendo una imagen personalizada de Docker para RabbitMQ

    Como mencionamos anteriormente, todos los servicios de nuestra plataforma de pruebas de carga se ejecutan dentro de contenedores Docker. Por lo tanto, tuvimos que construir una imagen personalizada de Docker para RabbitMQ con la biblioteca eflame incluida en el código del servidor. Creamos un repositorio de RabbitMQ-docker que hace que sea fácil construir una imagen de Docker con el código fuente de RabbitMQ modificado.

    Perfilando con eflame

    Una vez que tuvimos una imagen de Docker de RabbitMQ modificada con eflame incluido, pudimos ejecutar otra prueba de carga (las especificaciones eran las mismas que en la prueba anterior) y comenzar el perfilado real. Estos fueron los resultados:

    Realizamos varias mediciones y obtuvimos dos tipos de resultados como se presentan arriba. La principal diferencia entre estos gráficos se encuentra en la función rabbit_mgmt_util:run_run_augmentation/2 . ¿Qué significa esa diferencia?

    A partir de los resultados de las pruebas de carga anteriores y del análisis manual del código, sabemos que existen consultas lentas y rápidas. Las solicitudes lentas pueden tardar hasta veinte segundos, mientras que las rápidas solo tardan unos pocos segundos. Esto confirma el gráfico de tiempo de consulta anterior con un percentil del 50 de alrededor de 1,5 segundos, pero el 95 (y porcentajes más altos) que llegan hasta 20 segundos. Además, medimos manualmente el tiempo de ejecución de ambos casos utilizando timer:tc/3 y los resultados fueron consistentes.

    Esto sucede porque hay una caché en el plugin de Management. Cuando la caché es válida, las solicitudes se sirven mucho más rápido ya que los datos ya se han recopilado, pero cuando es inválida, es necesario recopilar toda la información necesaria.

    A pesar de que los gráficos tienen la misma longitud en la imagen, representan diferentes tiempos de ejecución (rápido vs lento). Por lo tanto, es difícil adivinar qué gráfico muestra qué consulta sin tomar realmente una medición. El primer gráfico muestra una consulta rápida mientras que el segundo muestra una consulta lenta. En el gráfico de consulta lenta rabbit_mgmt_util:augment/2 -> rabbit_mgmt_db:submit_cached/4 -> gen_server:call/3 -> … la pila tarda tanto tiempo porque la caché es inválida y se necesita recopilar datos nuevos. Entonces, ¿qué sucede cuando se recopilan los datos?

    Perfiles con fprof

    Podrías preguntar: “¿por qué no vemos las funciones de recopilación de datos en los gráficos de llama?” Esto sucede porque la caché se implementa como otro proceso de Erlang y la recopilación de datos ocurre dentro del proceso de caché. Hay una función gen_server:call/3 visible en los gráficos que hace una llamada al proceso de caché y espera una respuesta. Dependiendo del estado de la caché (válido o inválido), una respuesta puede volver rápidamente o lentamente.

    La recopilación de datos se implementa en la función rabbit_mgmt_db:list_queue_stats/3 que se invoca desde el proceso de caché. Naturalmente, deberíamos perfilar esa función. Probamos con eflame y después de varias docenas de minutos , este es el resultado que obtuvimos:

    eheap_alloc: Cannot allocate 42116020480 bytes of memory (of type "old_heap").
    

    El asignador de memoria del montón de Erlang intentó asignar 42 GB de memoria (de hecho, se necesitaba espacio para que el recolector de basura operara) y se bloqueó el servidor. Como eflame aprovecha el seguimiento de Erlang para generar gráficos de llama, probablemente se sobrecargó con la cantidad de eventos de seguimiento generados por la función rastreada. Ahí es donde entra en juego fprof.

    Según la documentación oficial de Erlang, fprof es:

    una herramienta de perfilado de tiempo que utiliza el seguimiento de archivo para un impacto mínimo en el rendimiento en tiempo de ejecución.

    Esto es muy cierto. La herramienta manejó la función de recopilación de datos sin problemas, aunque tardó varios minutos en producir el resultado. La salida fue bastante grande, así que solo se enumeran las líneas cruciales a continuación:

    (rabbit@rmq-test-rabbitmq-1)96> fprof:apply(rabbit_mgmt_db, list_queue_stats, [RA, B, 5000]).  
    ...
    (rabbit@rmq-test-rabbitmq-1)97> fprof:profile().  
    ...
    (rabbit@rmq-test-rabbitmq-1)98> fprof:analyse().  
    ...
    %                                       CNT        ACC       OWN  
    {[{{rabbit_mgmt_db,'-list_queue_stats/3-lc$^1/1-1-',4}, 803,391175.593,  105.666}],  
     { {rabbit_mgmt_db,queue_stats,3},              803,391175.593,  105.666},     %  
     [{{rabbit_mgmt_db,format_range,4},            3212,390985.427,   76.758},  
      {{rabbit_mgmt_db,pick_range,2},              3212,   58.047,   34.206},  
      {{erlang,'++',2},                            2407,   19.445,   19.445},  
      {{rabbit_mgmt_db,message_stats,1},            803,    7.040,    7.040}]}.  
    

    El resultado consiste en muchas de estas entradas. La función marcada con el carácter % es la que concierne a la entrada actual. Las funciones que se encuentran debajo son las que se llamaron desde la función marcada. La tercera columna ( ACC ) muestra el tiempo de ejecución total de la función marcada (tiempo de ejecución propio de la función y de los que la llaman) en milisegundos. Por ejemplo, en la entrada anterior, el tiempo de ejecución total de la función rabbit_mgmt_db:pick_range/2 es de 58.047 ms. Para obtener una explicación detallada de la salida de fprof, consulte la documentación oficial de fprof.

    La entrada anterior es la entrada de nivel superior que concierne a rabbit_mgmt_db:queue_stats/3 , que se llamó desde la función rastreada. Esa función gastó la mayor parte de su tiempo de ejecución en la función rabbit_mgmt_db:format_range/4. Podemos ir a una entrada que concierna a esa función y comprobar en qué gastó su tiempo de ejecución. De esta manera, podemos revisar la salida y encontrar posibles causas del problema de lentitud de la API de gestión.

    Al leer la salida de fprof de arriba hacia abajo, llegamos a esta entrada:

    {[{{exometer_slide,'-sum/5-anonymous-6-',7},   3713,364774.737,  206.874}],
     { {exometer_slide,to_normalized_list,6},      3713,364774.737,  206.874},     %
     [{{exometer_slide,create_normalized_lookup,4},3713,213922.287,   64.599}, %% SUSPICIOUS
      {{exometer_slide,'-to_normalized_list/6-lists^foldl/2-4-',3},3713,145165.626,   51.991}, %% SUSPICIOUS
      {{exometer_slide,to_list_from,3},            3713, 4518.772,  201.682},
      {{lists,seq,3},                              3713,  837.788,   35.720},
      {{erlang,'++',2},                            3712,   70.038,   70.038},
      {{exometer_slide,'-sum/5-anonymous-5-',1},   3713,   51.971,   25.739},
      {garbage_collect,                               1,    1.269,    1.269},
      {suspend,                                       2,    0.151,    0.000}]}.  
    

    La entrada se refiere a la función exometer_slide:to_normalized_list/6 que a su vez llamó a dos funciones “sospechosas” del mismo módulo. Profundizando encontramos esto:

    {[{{exometer_slide,'-create_normalized_lookup/4-anonymous-2-',5},347962,196916.209,35453.182},
      {{exometer_slide,'-sum/5-anonymous-4-',2},   356109,16625.240, 4471.993},
      {{orddict,update,4},                         20268881,    0.000,172352.980}],
     { {orddict,update,4},                         20972952,213541.449,212278.155},     %
     [{suspend,                                    9301,  682.033,    0.000},
      {{exometer_slide,'-sum/5-anonymous-3-',2},   31204,  420.574,  227.727},
      {garbage_collect,                              99,  160.687,  160.687},
      {{orddict,update,4},                         20268881,    0.000,172352.980}]}. 
    

    and

       {[{{exometer_slide,'-to_normalized_list/6-anonymous-5-',3},456669,133229.862, 3043.145},
      {{orddict,find,2},                           19369215,    0.000,129761.708}],
     { {orddict,find,2},                           19825884,133229.862,132804.853},     %
     [{suspend,                                    4754,  392.064,    0.000},
      {garbage_collect,                              22,   33.195,   33.195},
      {{orddict,find,2},                           19369215,    0.000,129761.708}]}.  
    

    Gran parte del tiempo de ejecución fue consumido por las funciones orddict:update/4 y orddict:find/2 . Ambas funciones combinadas representaron el 86% del tiempo total de ejecución.

    Esto nos llevó al módulo exometer_slide del plugin RabbitMQ Management Agent. Si se examina el módulo, se encontrarán todas las funciones mencionadas y las conexiones entre ellas.

    Decidimos cerrar la investigación en esta etapa porque este era claramente el problema. Ahora que hemos compartido nuestras reflexiones sobre el problema con la comunidad en este blog, quién sabe, tal vez encontraremos una nueva solución juntos.

    El efecto observador

    Hay una última cosa que es esencial considerar cuando se trata de depurar/observar sistemas: el efecto observador. El efecto observador es una teoría que afirma que si estamos monitoreando algún tipo de fenómeno, el proceso de observación cambia ese fenómeno.

    En nuestro ejemplo, utilizamos herramientas que se aprovechan del rastreo. El rastreo tiene un impacto en un sistema ya que genera, envía y procesa muchos eventos.

    Los tiempos de ejecución de las funciones mencionadas anteriormente aumentaron considerablemente cuando se llamaron con el perfilado habilitado. Las llamadas puras tomaron varios segundos mientras que las llamadas con el perfilado habilitado tomaron varios minutos. Sin embargo, la diferencia entre las consultas lentas y rápidas pareció permanecer sin cambios.

    El efecto observador no fue evaluado en el alcance del experimento descrito en esta publicación de blog.

    Una solución alternativa

    El problema puede ser resuelto de una manera ligeramente diferente. Pensemos por un momento si hay otra manera de obtener los nombres de las colas correspondientes a la cantidad de mensajes en ellas. Existe la función rabbit_amqqueue:emit_info_all/5 que nos permite recuperar la información exacta que nos interesa, directamente desde un proceso de cola. Podríamos utilizar esa API desde un plugin personalizado de RabbitMQ y exponer un punto final HTTP para enviar esos datos cuando se consulten.

    Convertimos esa idea en realidad y construimos un plugin de prueba de concepto llamado rabbitmq-queue-info que hace exactamente lo que se describe arriba. Incluso se realizó una prueba de carga del plugin (la especificación de la prueba fue exactamente la misma que la del plugin de gestión, como se mencionó anteriormente en el blog). Los resultados se muestran a continuación y hablan por sí solos:

    ¿Quieren más?

    ¿Quiere saber más sobre el rastreo en RabbitMQ, Erlang y Elixir? Consulte WombatOAM, un sistema intuitivo que facilita la supervisión y el mantenimiento de sus sistemas. Obtenga su prueba gratuita de 45 días de WombatOAM ahora.

    Apéndice

    La versión 3.7.9 de RabbitMQ se utilizó en todas las pruebas de carga mencionadas en esta publicación de blog. Un agradecimiento especial a Szymon Mentel y Andrzej Teleżyński por toda la ayuda con esa publicación.

    Nuestro trabajo con RabbitMQ.

    The post Cómo depurar tu RabbitMQ appeared first on Erlang Solutions .

    • chevron_right

      Ignite Realtime Blog: Release v1.1.0 of the MUC Real-Time Block List plugin for Openfire

      news.movim.eu / PlanetJabber · Saturday, 18 March, 2023 - 10:00 · 1 minute

    We are happy to announce the immediate availability of a new version of the MUC Real-Time Block List plugin for Openfire , our cross-platform real-time collaboration server based on the XMPP protocol! This plugin can help you moderate your chat rooms, especially when your service is part of a larger network of federated XMPP domains.

    From experience, the XMPP community has learned that bad actors tend to spam a wide range of public chat rooms on an equally wide range of different domains. Prior to the functionality provided by this plugin, the administrator of each MUC service had to manually adjust permissions, to keep unwanted entities out. With this new plugin, that process is automated.

    In this new release, several small bugs were fixed, and new features were introduced, including:

    • The plugin now, by default, uses a block list as maintained on https://xmppbl.org/
    • Support for blocking full domains (rather than just individual users) has been added
    • Block list entries no longer disappear over time

    The updated plugin should become available for download in your Openfire admin console in the course of the next few hours. Alternatively, you can download the plugin directly, from the plugin’s archive page .

    For other release announcements and news follow us on Twitter and Mastodon .

    1 post - 1 participant

    Read full topic

    • wifi_tethering open_in_new

      This post is public

      discourse.igniterealtime.org /t/release-v1-1-0-of-the-muc-real-time-block-list-plugin-for-openfire/92673

    • chevron_right

      Erlang Solutions: Here’s Why You Should Build Scalable Systems with Erlang

      news.movim.eu / PlanetJabber · Thursday, 16 March, 2023 - 10:00 · 5 minutes

    Building systems in the earlier days of the internet used to be pretty simple.

    While the system was admittedly pretty limited, the demand to scale past one or two servers wasn’t particularly high. But upon entering the 21st century, we saw large companies (think Amazon, Starbucks, Yahoo) and many more find the need to scale not just a few servers, but thousands. Even tens of thousands. Suddenly, the old-school system was impractical and nearly impossible to scale past one of two servers.

    The need for a system that offers scalability, flexibility and resilience had arrived. Enter Erlang- the powerful programming language designed for building highly scalable, fault-tolerant systems.

    Wondering what benefits there are to a scalable system like Erlang? Keep reading. We’ll be breaking down those very basics in this blog.

    A bit of background on Erlang

    But first, a bit of history of the Erlang language.

    Erlang was developed in the 1980s by Ericsson. Since then, it has been used to build large-scale distributed systems, such as telecom switching systems, online gaming platforms, and social networking sites.

    So, what is a scalable system?

    Before we start discussing scalable systems , let’s see what is really meant by the term ‘scaleable’.

    Measuring scalability is the ability to measure a system’s ability to increase or decrease in cost and performance, in response to the changes in a system’s demands.

    Now, it may seem obvious that an application being used by one user would require different levels of technology than one being used by a hundred. Yet, the reality is that there are still many businesses using technology that does not allow for this flexibility. This often leads to companies having to invest more money in creating software from scratch whenever they grow.

    As digital transformation drives accelerated business growth, businesses of all sizes need to be able to scale operations and adapt to their rapidly changing environments quickly. It’s no surprise that scalability has become an increasingly important factor when dealing with developing applications. Businesses have no choice but to be scalable, or they will face becoming overwhelmed when usage increases and will eventually become unable to meet the demands of a growing user base.

    A scalable computer language such as Erlang can write large new programmes and extend large existing ones relatively pain-free, depending on the complexity of the size of the programme it is trying to manage.

    Concurrency and parallelism in Erlang

    Erlang does a lot of things differently, one of those things being concurrency. When compared to most other programming languages that treat concurrency as an afterthought, Erlang builds in concurrency from the very base of the system.

    It was designed from the ground up to support concurrency and parallelism.

    Illustrating concurrency and parallelism on a 2-core CPU. Source: OpenClassrooms 2020. https://devopedia.org/concurrency-vs-parallelism

    Erlang’s lightweight processes, also known as actors, can execute in parallel, and they communicate with each other by exchanging messages. This message-passing model makes it easy to build highly concurrent systems that can handle a large number of users.

    Fault tolerance (Let it Crash)

    The philosophy behind Erlang is simply ‘Let it Crash.’

    Sound odd, right? Actually, letting it crash isn’t about crashing for the user or system.

    It’s about containing failure and letting Erlang clean it up.

    It knows that errors will happen, and things will break.

    Instead of trying to simply guard against those errors, Erlang has a built-in mechanism to handle those errors and failures.

    These mechanisms allow guarding against errors. So when a process crashes, it can restart automatically. In turn, the system restarts quickly and continues to operate smoothly.

    OTP

    The Open Telecom Platform (OTP) is a set of tools, frameworks and principles that are designed to guide and support the deployment of Erlang systems.

    OTP includes a wide range of components, such as a supervision tree, process registry, and message queues, which can be used to build complex distributed systems.

    Supervision tree example https://www.erlang.org/doc/design_principles/des_princ.htm

    Focusing on the supervision tree is a key hierarchical structuring model that is based on the idea of workers and supervisors, which makes it possible to design and programme fault-tolerant software.

    The workers are processes that perform computations meaning, they do the actual work. And supervisors are processes that monitor the behaviour of those workers. A supervisor can restart a worker if something goes wrong.

    Using OTP in your projects will help you to avoid accidental complexity.

    Distributed systems

    Erlang was designed for building distributed systems. It has built-in support for building systems that span multiple nodes. Erlang’s distribution mechanism allows processes to communicate with each other across the network, making it simple to build systems that scale horizontally.

    Erlang is high- performance

    Erlang is an incredibly high-performing language that can handle a large number of concurrent users and has great resilience over high task loads.

    Well known for its low latency, it is well-suited for building systems that require real-time processing and also has a small memory footprint, which enables it to run efficiently on low-end hardware.

    Hot Code Loading

    Erlang has a unique feature known as hot code loading, which enables developers to update their systems without shutting them down.

    Another way to think of it is that Hot Code Loading is the art of replacing an engine from a running car, without ever having to stop the car itself. It can update the code without causing any disruption to the service, meaning zero impact on users.

    This feature is particularly useful for building systems that need to be available 24/7.

    Scalability

    Last but not least, Erlang’s scalability is next to none. This language can be used to build systems that can handle millions of users. Erlang’s concurrency model and distributed architecture make it easy to build systems that can scale horizontally across multiple nodes, allowing developers to handle increasing loads, without sacrificing performance.

    To conclude

    Overall, Erlang is a great choice for building large-scale distributed systems that need to be highly available and performant. It handles concurrency and all its complexities with robustness and ease. But don’t just take our word for it. There are thousands of companies across the globe that have enlisted Erlang beyond its early days in telecoms.

    But don’t just take our word for it. Here are some of the clients who have felt the real-life impact of Erlang on their businesses .

    Fancy finding more out about Erlang? Check out our page .

    The post Here’s Why You Should Build Scalable Systems with Erlang appeared first on Erlang Solutions .

    • wifi_tethering open_in_new

      This post is public

      www.erlang-solutions.com /blog/heres-why-you-should-build-scalable-systems-with-erlang/

    • chevron_right

      Ignite Realtime Blog: Developing Openfire Efficient XML Interchange (EXI) functionality

      news.movim.eu / PlanetJabber · Tuesday, 14 March, 2023 - 19:39 · 1 minute

    We am excited to announce that a new plugin for the Openfire real time collaboration server is in the works! This plugin implements Efficient XML Interchange (EXI) functionality and provides an XMPP implementation of EXI as defined in XEP-0322 .

    Efficient XML Interchange (EXI) is a binary XML format for exchange of data on a computer network. It is one of the most prominent efforts to encode XML documents in a binary data format, rather than plain text. Using EXI format reduces the verbosity of XML documents as well as the cost of parsing.

    EXI is useful for:

    • a complete range of XML document sizes, from dozens of bytes to terabytes
    • reducing computational overhead to speed up parsing of compressed documents
    • increasing endurance of small devices by utilizing efficient decompression

    Read more about EXI in its Wikipedia article (where the above definition was taken from).

    The plugin that we’re developing today was first created by Javier Placencio, in 2013 and 2014. In 2023, that now dormant project was forked by the Ignite Realtime community.

    Work on the plugin is progressing steadily. Most of the core functionality is believed to be ready. In preparation for the official release of the plugin, we are looking for opportunities to perform interoperability testing. So far, testing has been done with our own mock client implementations. To be able to release a fully functional plugin, we’d like to test against implementations of other authors. Development builds of the plugin can be downloaded from the Openfire EXI plugin archive page .

    Are you interested in this? Please reach out to us on the Ignite Realtime Community , or stop by the open chat ! We would love to hear from you!

    For other release announcements and news follow us on Twitter and Mastodon .

    1 post - 1 participant

    Read full topic

    • wifi_tethering open_in_new

      This post is public

      discourse.igniterealtime.org /t/developing-openfire-efficient-xml-interchange-exi-functionality/92663

    • chevron_right

      Erlang Solutions: Presentamos el soporte de transmisión en RabbitMQ

      news.movim.eu / PlanetJabber · Monday, 13 March, 2023 - 19:33 · 11 minutes

    ¿Quiere saber más sobre el soporte de transmisión en RabbitMQ? Arnaud Cogoluègnes, ingeniero de personal de VMware, desglosa todo lo que hay que saber en la Cumbre RabbitMQ de 2021.

    En julio de 2021, se introdujeron streams a RabbitMQ, utilizando un nuevo protocolo extremadamente rápido que se puede utilizar junto con AMQP 0.9.1. Los streams ofrecen una forma más fácil de resolver varios problemas en RabbitMQ, incluyendo grandes fan-outs, replay y time travel, y grandes logs, todo con un rendimiento muy alto (1 millón de mensajes por segundo en un clúster de 3 nodos). Arnaud Cogoluègnes , Ingeniero de Staff en VMware , presentó los streams y cómo se utilizan mejor.

    Esta charla fue grabada en el RabbitMQ Summit 2021. La 4ta edición del RabbitMQ Summit se llevará a cabo como un evento híbrido, tanto en persona (en el lugar CodeNode en Londres) como virtual, el 16 de septiembre de 2022 y reunirá a algunas de las mayores empresas del mundo que utilizan RabbitMQ, todas en un solo lugar.

    Streams: Un Nuevo Tipo de Estructura de Datos en RabbitMQ

    Streams son una nueva estructura de datos en RabbitMQ que abren un mundo de posibilidades para nuevos casos de uso. Modelan un registro de solo agregado, lo que representa un gran cambio respecto a las colas tradicionales de RabbitMQ, ya que tienen semántica de consumidor no destructiva. Esto significa que cuando se leen mensajes de un Stream, no se eliminan, mientras que en las colas, cuando se lee un mensaje de una cola, se destruye. Este comportamiento reutilizable de RabbitMQ Streams se facilita mediante la estructura de registro de solo agregado.

    Text from the image:(Streams: a un nuevo tipo de estructura de datos en RabbitMQ)
    (Modela registros de solo anexar) (Persistente y replicado)(semántica de cliente no destructiva)(AMQP 0.9.1 y protocolo nuevo)

    RabbitMQ también introdujo un nuevo protocolo, el protocolo Stream, que permite un flujo de mensajes mucho más rápido. Sin embargo, también puedes acceder a Streams a través del protocolo AMQP 0.9.1 tradicional, que sigue siendo el protocolo más utilizado en RabbitMQ. También son accesibles a través de otros protocolos que RabbitMQ soporta, como MQTT y STOMP.

    Fortalezas de Streams

    Los Streams tienen fortalezas únicas que les permiten destacar en algunos casos de uso. Estas incluyen:

    Difusión masiva

    Cuando tienes varias aplicaciones en tu sistema que necesitan leer los mismos mensajes, tienes una arquitectura de difusión masiva. Los Streams son excelentes para las difusiones masivas, gracias a sus semánticas de consumo no destructivas, eliminando la necesidad de copiar el mensaje dentro de RabbitMQ tantas veces como haya consumidores.

    Reproducción y viaje en el tiempo

    Los Streams también ofrecen capacidades de reproducción y viaje en el tiempo. Los consumidores pueden adjuntarse en cualquier lugar de un Stream, utilizando un desplazamiento absoluto o una marca de tiempo, y pueden leer y volver a leer los mismos datos tantas veces como sea necesario.

    Rendimiento

    Gracias al nuevo protocolo de stream, los streams tienen el potencial de ser significativamente más rápidos que las colas tradicionales. Si necesitas un alto rendimiento o estás trabajando con mensajes grandes, los streams a menudo pueden ser una opción adecuada.

    Mensajes grandes

    Los Streams también son buenos para grandes registros. Los mensajes en los streams siempre son persistentes en el sistema de archivos, y los mensajes no permanecen en la memoria por mucho tiempo. Al consumirse, se utiliza la caché de archivos del sistema operativo para permitir un flujo de mensajes rápido.

    RabbitMQ also introduced a new protocol, the Stream protocol, which allows much faster message flow, however, you can access Streams through the traditional AMQP 0.9.1 protocol as well, which remains the most used protocol in RabbitMQ. They are also accessible through the other protocols that RabbitMQ supports, such as MQTT and STOMP.

    Text from the image:(Grandes fan-outs) (Repetición/tiempo de viaje)(Alto rendimiento)(Grandes registros)

    La Abstracción del Log

    Un stream es inmutable, puedes añadir mensajes, pero una vez que un mensaje ha entrado en el stream, no se puede eliminar. Esto hace que la abstracción del log del stream sea una estructura de datos bastante simple en comparación con las colas donde los mensajes siempre se añaden y se eliminan. Esto nos lleva a otro concepto importante, el offset. El offset es simplemente un índice técnico de un mensaje dentro del stream, o una marca de tiempo. Los consumidores pueden indicar a RabbitMQ que empiece a leer desde un offset en lugar del principio del stream. Esto permite una fácil reproducción y viaje en el tiempo de los mensajes. Los consumidores también pueden delegar la responsabilidad de seguimiento del offset a RabbitMQ.

    Text from the image:
    (La abstracción de registros)
    (Un modelo de secuencias y un registro de solo anexar)

    (Estructura de datos FIFO)
    (Lectura no destructiva)

    (Mensaje más viejo) (Compensar)(Último mensaje)(Proximo mensaje iría aquí en su lugar)

    Podemos tener cualquier cantidad de consumidores en un stream, no compiten entre sí, una aplicación consumidora no robará mensajes de otras aplicaciones, y la misma aplicación puede leer el flujo de mensajes muchas veces.

    Las colas pueden almacenar mensajes en memoria o en disco, pueden estar en un solo nodo o estar replicadas, los streams son persistentes y replicados en todo momento. Cuando creamos un stream, tendrá un líder ubicado en un nodo y réplicas en otros nodos. Las réplicas seguirán al líder y sincronizarán los datos. El líder es el único que puede crear operaciones de escritura y las réplicas solo se utilizarán para servir a los consumidores.

    Colas de RabbitMQ vs. Streams

    Los streams están aquí para complementar las colas y ampliar los casos de uso de RabbitMQ. Las colas tradicionales siguen siendo la mejor herramienta para los casos de uso más comunes en RabbitMQ, pero tienen sus limitaciones, hay momentos en los que no son la mejor opción.

    Los streams son, al igual que las colas, una estructura de datos FIFO, es decir, el mensaje más antiguo publicado se leerá primero. Proporcionar un desplazamiento permite al cliente omitir el comienzo del flujo, pero los mensajes se leerán en el orden de publicación.

    En RabbitMQ, tiene una cola tradicional con un par de mensajes y una aplicación consumidora. Después de registrar el consumidor, el intermediario comenzará a despachar mensajes al cliente y la aplicación puede comenzar a procesarlos.

    Cuando, en este punto, el mensaje está en un punto importante de su vida útil, está presente en el lado del remitente y también en el lado del consumidor. El intermediario todavía necesita preocuparse por el mensaje porque puede ser rechazado y debe saber que aún no se ha reconocido. Después de que la aplicación termine de procesar el mensaje, puede reconocerlo y, a partir de este momento, el intermediario puede deshacerse del mensaje y considerarlo procesado. Esto es lo que podemos llamar consumo destructivo, y es el comportamiento de las colas clásicas y de cuórum. Al usar Streams, el mensaje permanece en el Stream mientras la política de retención lo permita.

    Implementar configuraciones de gran difusión masiva con RabbitMQ no era óptimo antes de Streams. Cuando entra un mensaje, va a un intercambio y se enruta a una cola. Si desea que otra aplicación procese los mensajes, debe crear una nueva cola, vincular la cola al intercambio y comenzar a consumir. Este proceso crea una copia del mensaje para cada aplicación, y si necesita que otra aplicación procese los mismos mensajes, debe repetir el proceso; entonces otra cola, un nuevo enlace, un nuevo consumidor y una nueva copia del mensaje.

    Este método funciona y se ha utilizado durante años, pero no escala de manera elegante cuando se tienen muchas aplicaciones consumidoras. Los streams proporcionan una mejor manera de implementar esto, ya que los mensajes pueden ser leídos por cada consumidor por separado, en orden, desde el Stream.

    Rendimiento de los streams de RabbitMQ usando AMQP y el protocolo de Stream

    Como se explica en la charla, hubo un mayor rendimiento con Streams en comparación con las colas de cuórum.

    Obtuvieron alrededor de 40,000 mensajes por segundo con las Colas Quórum y 64,000 mensajes por segundo con Streams. Esto se debe a que Streams son una estructura de datos más simple que las Colas Quórum, ya que no tienen que lidiar con cosas complicadas como la confirmación de mensajes, mensajes rechazados o reencolado.

    Text from the image:
    (Streams en AMQP)
    (Cluster de 3 nodos (instancias c2-standard-16))
    (Tarifas de publicación)
    (mensajes/segundos)

    (Colas Quorum)(Stream en AMQP)

    Las colas de Quorum siguen siendo colas replicadas y persistentes de vanguardia, mientras que las Streams son para otros casos de uso. Al usar el protocolo Stream dedicado, se pueden lograr tasas de transferencia de un millón de mensajes por segundo.

    Text from the image: (Protocolo Stream)
    (Cluster de 3 nodos (instancias c2-standard-16))
    (Tarifas de publicación)
    (mensajes/segundos)
    (Colas Quorum)
    (Stream en AMQP)
    (Stream con protocolo de stream)

    El Protocolo Stream ha sido diseñado teniendo en cuenta el rendimiento y utiliza técnicas de bajo nivel como la API libC sendfile, la caché de página del sistema operativo y el agrupamiento, lo que lo hace más rápido que las colas AMQP.

    El plugin RabbitMQ Stream y los clientes

    Los Streams están disponibles a través de un nuevo plugin en la distribución principal. Cuando está activado, RabbitMQ comenzará a escuchar en un nuevo puerto que puede ser utilizado por los clientes que comprenden el Protocolo Stream. Está integrado con la infraestructura existente en RabbitMQ, como la interfaz de gestión, la API REST y Prometheus.

    Text from the image: (Los Streams son también accesibles a través de un nuevo protocolo)
    (Rápido) (Complemento en la distribución principal)(integración de gestión)

    Hay un cliente dedicado en Java y Go que utiliza este nuevo protocolo de flujo. El cliente de Java es la implementación de referencia. También está disponible una herramienta de prueba de rendimiento. Los clientes para otros lenguajes también son desarrollados activamente por la comunidad y el equipo central.

    El protocolo de flujo es un poco más simple que AMQP; no hay enrutamiento; simplemente se publica en un flujo, no hay intercambio involucrado, y se consume de un flujo como de una cola. No se necesita lógica para decidir dónde se debe dirigir el mensaje. Cuando publicas un mensaje desde tus aplicaciones cliente, este va a la red y casi directamente al almacenamiento.

    Existe una excelente interoperabilidad entre los flujos y el resto de RabbitMQ. Los mensajes se pueden consumir desde una aplicación cliente AMQP 0.9.1 y también funciona en sentido contrario.

    Ejemplo de caso de uso para interoperabilidad:

    Las colas y los flujos viven en el mismo espacio de nombres en RabbitMQ, por lo que se puede especificar el nombre del flujo del que se desea consumir utilizando los clientes AMQP habituales y mediante el parámetro x-stream-offset para basicConsume.

    Es muy fácil publicar con clientes AMQP porque es lo mismo que con las colas, se publica en un intercambio.

    Text from the image: (Agregar Stream para Analiticas)
    (Editor)(Cola) (Procesando AMER)
    (Editor)(Cola) (Procesando EMEA)
    (Editor)(Cola) (Procesando APAC)
    (Cola) (Analiticas globales)
    ((Posibilidad) editores multiprotocolos)

    Arriba se muestra un ejemplo de cómo se puede imaginar el uso de streams. Se tiene un publicador que publica mensajes en un exchange y, según la clave de enrutamiento de los mensajes, se enrutan a diferentes colas. Por lo tanto, se tiene una cola para cada región del mundo. Por ejemplo, se tiene una cola para las Américas, una para Europa, una para Asia y una para la sede. Se tiene una aplicación consumidora dedicada que realizará un procesamiento específico para cada región.

    Si se actualiza a RabbitMQ 3.9 o posterior, se puede simplemente crear un stream, vincularlo al exchange con un comodín para que todos los mensajes se enruten a las colas pero el stream reciba todos los mensajes. Luego se puede dirigir una aplicación que utiliza el Protocolo Stream a este stream y podemos imaginar que esta aplicación realizará análisis mundiales todos los días sin siquiera leer el stream muy rápido. Así es como podemos imaginar que los streams se ajustan a las aplicaciones existentes.

    Garantías para RabbitMQ Streams

    Los streams admiten entrega al menos una vez, ya que admiten un mecanismo similar a AMQP Publish Confirms. También hay un mecanismo de deduplicación, el agente filtra los mensajes duplicados según el número de secuencia de publicación, como una clave en una base de datos o un número de línea en un archivo.

    Text from the image: (Garantías)                              (Mensajes de deduplicación)                (Control de flujo)
    (Al menos uno)                        (publicando)
    (Sin pérdida de mensajes)

    En ambos lados, tenemos control de flujo, por lo que se bloquearán las conexiones TCP de los editores rápidos. El corredor solo enviará mensajes al cliente cuando esté listo para aceptarlos.

    Resumen

    Text from the image: (Streams: una nueva estructura de tipo de registro replicada y persistente en RabbitMQ)
    (Desbloquear los nuevos escenarios con RabbitMQ)
    (Grandes fan-outs) (Repetición/tiempo de viaje)(Alto rendimiento)(Grandes registros)

    (Pruebalo)

    Los Streams son una nueva estructura de datos replicada y persistente en RabbitMQ, que modelan un registro de solo anexión. Son buenos para distribución masiva, soportan funciones de reproducción y viaje en el tiempo, son adecuados para escenarios de alto rendimiento y para grandes registros. Almacenan sus datos en el sistema de archivos y nunca en memoria.

    Si crees que los Streams o RabbitMQ podrían ser útiles para ti pero no sabes por dónde empezar, habla con nuestros expertos , siempre estamos dispuestos a ayudar. Si quieres ver las últimas características y estudios de casos del mundo de RabbitMQ, únete a nosotros en RabbitMQ Summit 2022.

    The post Presentamos el soporte de transmisión en RabbitMQ appeared first on Erlang Solutions .

    • wifi_tethering open_in_new

      This post is public

      www.erlang-solutions.com /blog/presentamos-el-soporte-de-transmision-en-rabbitmq/