Make sure you commit anything you want to end up in the KDE Gear 25.04
releases to them
Next Dates
March 13 2025: 25.04 Freeze and Beta (25.03.80) tag & release
March 27, 2025: 25.04 RC (25.03.90) Tagging and Release
April 10, 2025: 25.04 Tagging
April 17, 2025: 25.04 Release
https://community.kde.org/Schedules/KDE_Gear_25.04_Schedule
So, after my last blog post, I ended up taking another few months to get the fonts branch just right, but now we have font resource that can be tagged, filtered and searched upon.
After that, I needed my next text editing project to be a bit more manageable. Given that I had already made some head start on it at the beginning of last year, I continued with UI for the OpenType features.
Usually, OpenType features are explained away as “that’s for ligatures and stuff”, which, while not incorrect, is maybe a little simple. “It enables advanced typographic features” is a little more correct, but it makes OpenType feature support sound less rudimentary than it really is these days.
What might be more clear is to think of a font file as a mapping between input characters and their glyphs. This is fine for simple Latin. But what if you want to show Arabic connected? Well, then you need a second table to keep track of the glyphs for start, middle and end of a word, and substitute the correct glyph as necessary.
Or maybe you want to have kerning, so that A and V nest into each other nicely. Another table for that then, that keeps track of the position adjustment between two consecutive glyphs.
How about small caps? Substitution table. Dynamically placing diacritics? Positioning table. Cyrillic has different glyph traditions in Serbia and Bulgaria, similarly for Han script use in East-Asia: Substitution table. Han script is usually typeset in mono space, but when two brackets follow one another, they often have extraneous white space that should be removed: Positioning table. These are all OpenType features.
The more you look into it, the more it becomes clear that if you want your text layout to support more than plain English, you will need to allow these extra tables to be used and read through. This is in short what Harfbuzz does for us. You can enable and disable these features by taking their name (A 4 letter tag), and indicating to Harfbuzz you want them enabled for a given section by sending a number (0 for off, 1 for on… and 2, 3, 4, […] 256 for indicating which alternate you’d like if the font has multiple alternates available for that feature).
Some of these features are enabled by default, and the basic text layout will process them just fine. However, others are optional, and within CSS all of them can be turned off and on at the typesetters’ wish. Which brings us to the widgets.
There’s two types of CSS opentype toggles. The first of these are the font-variants, which have a somewhat confusing name in this day and age of OpenType variable fonts, but at the time they were named font-variants were limited to Small Caps, and expected to be separate font files.
The font-variant features more of a guidance suggestion, meaning that you turn them on or off for a piece of text, unrelated to whether the current font actually supports the feature in question. The idea being that you could set a different font elsewhere in the text, and that if this font supports those features, they could be controlled this way.
This means that the UI for these features is somewhat straight forward and unopinionated, being a collection of drop-downs and check boxes. I renamed them “glyphs:” in the UI to avoid confusion with Variable fonts, which is also possible because the majority of them represented which glyphs are being used.
font-variants only cover the most common features, and as of writing, there’s over 120 registered OpenType feature tags. CSS allows controlling this via the second type of properly “font-feature-settings”. A property that wants you to be very specific about which tag you want to enable, and whether you want to have it not just enabled, but also which sub index of the feature you would like to enable.
Now, there’s a bit of a problem here: 120 features is a bit much. Furthermore, two of those registered features, Character Variants and Stylistic Sets, are registered 99 and 20 times respectively, meaning the total is closer to over 230 features. And, further furthermore, fonts may have custom OpenType feature tags.
And that’s not the only problem: Access All Alternates, Stylistic Alternates and Stylistic Sets are very common features, but the way they are configured in CSS as a font-variant feature is somewhat complex, and to have each manually enabled inside the OpenType Features widget is going to feel very clunky for artists.
For these reasons, I ended up building two controls for this CSS property. A main widget for the text property docker that allows artists to enable and disable any OpenType feature that can be found inside the font, and the second being a glyph palette, that allows artists to select alternate glyphs.
The glyph palette was actually made first, so lets go over that. It is in effect a character selection map that allows artists to select alternates to the current glyph or to any glyph inside the font. Filtering can be done with Unicode blocks and search.
It uses KoFontGlyphModel, a QAbstractItemModel, that collects all the available characters inside the font, as well as their Unicode variations (Unicode variations are an extra code point added behind a character to indicate it needs to be a specific version of that glyph. It’s main use is to ensure people’s names are written with the correct glyph variant).
It then takes the locale of the text, and uses those to go over the OpenType tables with Harfbuzz. The locale of the text, or “text language” is necessary because some OpenType features are only available for certain locales. Furthermore, the aforementioned Character Variants and Stylistic Sets may have been named inside the font, meaning that it also takes the interface languages to get the correct name for the current localization of Krita.
Of course, using these alternate names means we need default names first. Which is why I also spend some time on creating a factory where each known OpenType tag is stored with its proper name from the official registry and a hand written summary of the feature for the tool tip. These can now be localized.
Then when we have the feature, we go over the glyphs marked by the table and if that glyph coheres with a Unicode code point, add the table as a potential sub glyph for that Unicode value.
Now here another problem rears its head: We need to know which glyph coheres with which Unicode code point, and while for basic values that isn’t a problem, it is when decomposition comes into play.
Decomposition in this case, is a feature that allows for replacing a given glyph with two other glyphs. A reverse ligature, if you will. It is frequently used to match Unicode decomposition: Ä according to Unicode can be decomposed into A (U+0041) and ◌̈ (U+0308, combining diaeresis). So then, the glyph for Ä can be decomposed into those two glyphs. Afterwards, OpenType allows positioning that diaeresis accurately with the mark positioning feature. This is useful, because we can then take that A glyph, and use things like the Small Caps feature, or a Stylistic Set to turn it into a small A or a decorative A, and as long as these alternate glyphs have been configured for mark positioning, they’ll by themselves support Ä.
So that’s pretty neat. But the problem is that Harfbuzz doesn’t provide enough information for me to discover how a glyph gets decomposed. Meaning that for fonts that are structured this way, I can’t tell whether style sets or the like can be applied to these glyphs, so these don’t show up in the character map. I have a similar problem with ligatures, but that is also compounded by having trouble with the user interface.
For the glyph palette, the way you use it is either by using the glyph alternates, where double clicking one will replace the current grapheme (that’s a set of Unicode values that is often treated as the smallest editable chunk of text) with one that either has the appropriate Unicode variation selectors attached, or one that has the appropriate OpenType features enabled.
The other option is to use the character map, which is much like character maps in other software, allowing you to scroll over all available Unicode values in a font, and sorting them by Unicode block, or searching them. Clicking on a glyph with variants ops out a context menu with the glyph alternates.
The glyph palette itself is written with QML, but because the rest of Krita is not in QML, it is embedded into a QQuickWidget, that is inside a QDialog, which in turn means the context menu needed to be inside a QQuickWidget inside a QPopup, because QQuickWidget will clip any QML pop-up items. QML side, we use DelegateModel to show the child indices of a given character map index.
I’m not sure yet how ligatures would be handled here, maybe list them for each glyph, or maybe have a separate model that shows up underneath the glyph alternates and only shows for the current text. There’s also the fact that stylistic alts and the like can apply to ligatures, so that’s another thing to consider. A similar issue is with emoji zero-width-joiner sequences. This is stuff like “fireman + woman + Fitzpatrick skin tone modifier 5” = 
. This is typically implemented as a kind of ligature as well, and while Unicode keeps a list of these, I’d prefer to get them from the font.
For the “OpenType features” control in the text properties docker, we reuse the glyph model, but this time its only to figure out which features are available in the font. Because CSS allows for font fallback, we only do this for the first font, but also allow setting any other officially registered OpenType feature on or off. It also shows a sample for the given feature. This widget is mostly useful for the stylistic sets and the positioning features.
Now, setting up the glyph model can get quite slow, so some trade-offs were established:
One of the things that struck me when writing the original svg text layout post a few years back is that a decade ago, you’d really boast about your OpenType support, but nowadays, OpenType support is so rudimentary, it didn’t make sense to dwell on it in that post. This might also be a consequence by how easy Harfbuzz makes using OpenType these days.
That meant I really wanted to get the UI for this feature nice. There was a big post over a decade ago by a UI designer doing UI for free software, where he went into extensive detail about how most software implementing controls for OpenType features is really bad at communicating whether the feature does anything. I think I managed to somewhat get that part working right.
Still, the glyph palette could use more love, and I really need to sit down for the whole ligature UI issue. I’m pretty happy with it none the less, and it is very hackable, meaning that it doesn’t necessarily need to be me personally improving it.
I do need to really get going on that language selector though…
The code for retrieving the OpenType tables was largely based on Inkscape’s, and then extended. Inkscape doesn’t test on language, and only tests for substitution features, while we test on both substitution and positioning features. Similarly, Inkscape’s was written in a time when Harfbuzz could only give information about whether a feature could be turned only on or off, but not whether it had multiple alternates, so it is not yet able to do that.
Of interest is that Inkscape does show a few ligatures, but the only reason those are visible is that there’s a handful of ligatures that are encoded into Unicode in the “Alphabetic Presentation Forms” block. Fonts that implement ligatures tend to also setup these Unicode values, but this is not self-evident, which is why I’d prefer not doing this.
(As a random factoid: Adobe’s Smart Quote feature will use these Unicode encoded ligatures when the font isn’t able to provide them via OpenType.)
I did manage to get ligature samples by simply testing every combination of glyphs that Harfbuzz could tell me were probably relevant to a given table, but this was slow, leading to a 5~ second initialization time on a feature heavy font like Junicode. Maybe the glyph model code can be at some point modified to allow incremental loading, though that wouldn’t provide me a quick sample text in the text properties docker…
I feel I should probably mention that OpenType isn’t the only technology that provides shaping. Apple’s Advanced Typography Tables (ATT) and the Graphite shaping language are existing alternatives, but OpenType is far more popular than either, and the CSS working group doesn’t give much guidance on how to use anything but OpenType.
Qt currently has two UI toolkits: QML and QWidget. The former uses the terminology “Item” instead of “Widget” to refer to UI controls. I find this somewhat difficult to get used to, so when I don’t prepend a widget name with Q, assume that I mean a generic UI control. I think most people never even consider what the different UI bits are called, so usually it isn’t a problem.
Just a quick update: Recently, you might have heard that GTK 4 added support for the cursor-shape-v1 protocol on Wayland. The main advantage of the cursor-shape-v1 protocol is that it ensures consistent cursor look between apps. In Plasma, additional perks come with it, for example the cursor can look more crisp with fractional scale factors because SVG cursors are used. We (KDE) took a shot at backporting the cursor shape protocol support to the GTK 3 branch and, as of this moment, it’s already merged 
. This means that you should experience fewer cursor issues in applications running on Wayland that still use GTK 3, mainly Firefox.
I would like to express many thanks to Max Qian for starting the work on adding support for the cursor-shape-v1 protocol, and Matthias Clasen for pushing it over the finish line and reviewing our backport MR.
Qt’s 3D offering is changing, so we decided to look at different options for rendering 3D content in Qt.
Continue reading 3D Rendering Solutions in Qt – an Overview at basysKom GmbH.
Last week, part of the Kdenlive core team met in Amsterdam for a short sprint, the highlight of which was a visit to the Blender Foundation.
Francesco Siddi, COO at Blender, provided us with a rare insight into Blender’s history and precious advice about product management for Kdenlive – we hope to implement some of these advices soon.
As the meeting took place on a Friday afternoon, we also had the opportunity to attend their “Weekly”, which is an open session where artists and developers share their progress of the past week on various Blender related projects.
So thanks again to Francesco and everyone at the Blender Foundation for their hospitality.
On the next day, we discussed a few topics, including:
Welcome to a new issue of "This Week in KDE Apps"! Every week we cover as much as possible of what's happening in the world of KDE apps. This time again a bit delayed. If you are a non technical person and are looking at a way to contribute to KDE, you can help editing "This Week in KDE Apps" would be very much welcome. Just join our Matrix chat.
This week we have some big changes in Krita, a redesign in SystemDGenie and a new, more compact statusbar for Dolphin.
We dropped the Qt5 support and moved to Qt6 (Tuomas Nurmi, Link).
Dolphin now uses a more compact statusbar by default (Akseli Lahtinen, 25.04.0. Link).
When in selection mode, Dolphin now has a special keyboard navigation mode. You can read all about this feature in detail in the merge request description (Felix Ernst, 25.04.0. Link).
We fixed various usability issues and recent regressions (Bart De Vries, 25.04.0. Link 1, link 2, link 3, ...).
We improved support for DAP (the generic protocol for debuggers) (Waqar Ahmed, 25.04.0 Link), and sped up KWrite's startup time by not loading a MIME database when just querying the icon for text/plain file. (Kai Uwe Broulik, 25.04.0. Link)
We removed the Decrypt/Verify all files in folder menu item in the Dolphin context menu as it was never implemented (Tobias Fella, 25.04.0. Link).
The Save As dialog now remembers where a file was last downloaded and will open that directory. Note that the last location is only remembered for the duration of the Konqueror window (Stefano Crocco, 25.04.0. Link).
We added a glyph palette to choose between alternates or variation of a given glyph, as well as a character map of a given font (Wolthera van Hövell, Link).
And implemented the edition of the CSS Font Variants in the text properties docker (Wolthera van Hövell, Link).
Krita now compiles with Qt6 on Windows (Dmitry Kazakov, Link).
We added a new extension "Mutator". This new extension provides a docker which adds brush variations through action-invoked settings randomization (Emmet O'Neill, Link). We also added global pen tilt direction offset which is helpful to make brushes feel the same for right- and left-handed users (Maciej Jesionowski. Link). Another brush related improvement is that their smoothness is now also affected by the speed (killy |0veufOrever, Link).
We improved the link text in the KCM user interface (Robert Kratky - first contribution 🚀, 0.11.0. Link).
Long pressing has been disabled on non-touchscreen devices (Joshua Goins, 25.04.0. Link), and we improved the usability of the account menu by giving it a proper button (Joshua Goins, 25.04.0. Link).
We have improved the error handling entering a bad passphrase for a digital signature (Sune Vuorela, 25.04.0. Link) and made the overprint preview setting a combobox that gives you the option to choose between "Always", "never" and "Automatic", which is similar to Acrobat Reader. The "Automatic" value depends on the value of HasVisibleOverprint in the PDF metadata (Kevin Ottens, 25.04.0. Link).
SystemDGenie was ported to a more "frameless" interface and the statusbar was removed (Thomas Duckworth. Link 1 and link 2).
SystemDGenie shows unloaded and inactive units by default (Thomas Duckworth. Link) and the startup time was sped up by fetching the list of sessions and units asynchronously (Carl Schwan. Link).
This blog only covers the tip of the iceberg! If you’re hungry for more, check out Nate's blog about Plasma and be sure not to miss his This Week in Plasma series, where every Saturday he covers all the work being put into KDE's Plasma desktop environment.
For a complete overview of what's going on, visit KDE's Planet, where you can find all KDE news unfiltered directly from our contributors.
The KDE organization has become important in the world, and your time and contributions have helped us get there. As we grow, we're going to need your support for KDE to become sustainable.
You can help KDE by becoming an active community member and getting involved. Each contributor makes a huge difference in KDE — you are not a number or a cog in a machine! You don’t have to be a programmer either. There are many things you can do: you can help hunt and confirm bugs, even maybe solve them; contribute designs for wallpapers, web pages, icons and app interfaces; translate messages and menu items into your own language; promote KDE in your local community; and a ton more things.
You can also help us by donating. Any monetary contribution, however small, will help us cover operational costs, salaries, travel expenses for contributors and in general just keep KDE bringing Free Software to the world.
To get your application mentioned here, please ping us in invent or in Matrix.
Edit 2025-03-17: applications for this position are closed.
KDE e.V., the non-profit organization supporting the KDE community, is looking for a contractor to improve KDE’s Plasma desktop environment in ways that support user acquisition through growth into new hardware and software markets. The Plasma software engineer will address defects and missing features that are barriers to these objectives. Please see the full job listing for more details about this opportunity. We are looking forward to your application.
The Amarok Development Squad is happy to announce the immediate availability of Amarok 3.2.2, the second bugfix release for Amarok 3.2 "Punkadiddle"!
3.2.2 features some minor bugfixes, and improvements for building Amarok on non-UNIX systems and without X11 support. Additionally, a 16-year-old feature request has been fulfilled. Concluding years of Qt5 porting and polishing work, Amarok 3.2.2 is likely to be the last version with Qt5/KF5 support, and it should provide a nice and stable music player experience for users on various systems and distributions. The development in git, on the other hand, will soon switch the default configuration to Qt6/KF6, and focus for the next 3.3 series will be to ensure that everything functions nicely with the new Qt version.
In addition to source code, Amarok is available for installation from many distributions' package repositories, which are likely to get updated to 3.2.2 soon, as well as the flatpak available on flathub.
You can find the tarball package on download.kde.org and it has been signed with Tuomas Nurmi's GPG key.
I’m Rishav Ray Chaudhury, a third year Electrical Engineering undergrad from India. I have participated in Season of KDE this year. I am tasked with implementing a variant of the Mancala board game called Kalah under the guidance of João Gouveia and Benson Muite.
As with any aspiring developer, I too started out with game development. A couple of my friends and I started to make games using C# and the Unity framework in my first year of University. The feeling of making games that my friends enjoyed was exhilarating to say the least. This was also around the same time that I learned about open-source software. Apparently, a lot of software that developers frequently use, was developed by a group of passionate programmers. I tried contributing to some repos but was unsuccessful, mainly because I was completely unfamiliar with the projects. That all changed when I started using Arch Linux. For Arch, the desktop environment that I chose was KDE Plasma. After using it for some time, I came to know about Season of KDE and finally took the initiative to start contributing to software that I frequently used. Of the projects, the game development project was the one that caught my eye and now here I am developing a game for it.
Modern distributed systems need to process massive amounts of data efficiently while maintaining strict ordering guarantees. This is especially challenging when scaling horizontally across multiple nodes. How do we ensure messages from specific sources are processed in order while still taking advantage of parallelism and fault tolerance?
Elixir, with its robust concurrency model and distributed computing capabilities, is well-suited for solving this problem. In this article, we’ll build a scalable, distributed message pipeline that:
Many modern applications require processing large volumes of data while preserving message order from individual sources. Consider, for example, IoT systems where sensor readings must be processed in sequence, or multi-tenant applications where each tenant’s data requires sequential processing.
The solution we’ll build addresses these requirements by treating each RabbitMQ queue as an ordered data source.
Let’s explore how to design this system using Elixir’s distributed computing libraries: Broadway, Horde, and libcluster.
The system consists of multiple Elixir nodes forming a distributed cluster. Each node runs one or more Broadway pipelines to process messages from RabbitMQ queues and forward them to Google Cloud PubSub. To maintain message ordering, each queue has exactly one pipeline instance running across the cluster at any time. If a node fails the system must redistribute its pipelines to other nodes automatically, and if a new node joins the cluster then the existing pipelines should be redistributed to ensure a balanced load.
Elixir natively supports the ability to cluster multiple nodes together so that processes and distributed components within the cluster can communicate seamlessly. We will employ the libcluster library since it provides several strategies to automatize cluster formation and healing.
For the data pipelines, the Broadway library provides a great framework to support multi-stage data processing while handling back-pressure, batching, fault tolerance and other good features.
To correctly maintain the distribution of data pipelines across the Elixir nodes, the Horde library comes to the rescue by providing the building blocks we need: a distributed supervisor that we can use to distribute and maintain healthy pipelines on the nodes, and a distributed registry that we use directly to track which pipelines exist and on which nodes they are.
Finally, a PipelineManager component will take care of monitoring RabbitMQ for new queues and starting/stopping corresponding pipelines dynamically across the cluster.
Let’s initiate a new Elixir app with a supervision tree.
mix new message_pipeline --sup
First, we’ll need to add our library dependencies in mix.exs and run mix deps.get:
defmodule MessagePipeline.MixProject do
use Mix.Project
def project do
[
app: :message_pipeline,
version: "0.1.0",
elixir: "~> 1.17",
start_permanent: Mix.env() == :prod,
deps: deps()
]
end
def application do
[
extra_applications: [:logger],
mod: {MessagePipeline.Application, []}
]
end
defp deps do
[
{:libcluster, "~> 3.3"},
{:broadway, "~> 1.0"},
{:broadway_rabbitmq, "~> 0.7"},
{:google_api_pub_sub, "~> 0.34"},
{:goth, "~> 1.4"},
{:tesla, "~> 1.12"},
{:jason, "~> 1.4"},
{:amqp, "~> 4.0"}
]
end
end
Since we just need to publish messages to Google Cloud PubSub and the API is straightforward, we don’t need fancy libraries here.
Let’s start by adding Goth to the supervision tree to manage Google’s service account credentials.
defmodule MessagePipeline.Application do
use Application
def start(_type, _args) do
credentials =
"GOOGLE_APPLICATION_CREDENTIALS_PATH"
|> System.fetch_env!()
|> File.read!()
|> Jason.decode!()
children = [
{Goth, name: MessagePipeline.Goth, source: {:service_account, credentials}},
# Other children...
]
Supervisor.start_link(children, strategy: :one_for_one)
end
end
And here’s our HTTP client to publish messages to Google Cloud PubSub
defmodule MessagePipeline.GooglePubsub do
@google_pubsub_base_url "https://pubsub.googleapis.com"
def publish_messages(messages) when is_list(messages) do
project_id = System.fetch_env!("GOOGLE_PUBSUB_PROJECT_ID")
topic_name = System.fetch_env!("GOOGLE_PUBSUB_TOPIC")
topic = "projects/#{project_id}/topics/#{topic_name}"
request = %{
messages: Enum.map(messages, &%{data: Base.encode64(&1)})
}
with {:ok, auth_token} <- generate_auth_token(),
client = client(auth_token),
{:ok, response} <- Tesla.post(client, "/v1/#{topic}:publish", request) do
%{body: %{"messageIds" => message_ids}} = response
{:ok, message_ids}
end
end
defp client(auth_token) do
middleware = [
{Tesla.Middleware.BaseUrl, @google_pubsub_base_url},
Tesla.Middleware.JSON,
{Tesla.Middleware.Headers, [{"Authorization", "Bearer " <> auth_token}]}
]
Tesla.client(middleware)
end
defp generate_auth_token do
with {:ok, %{token: token}} <- Goth.fetch(MessagePipeline.Goth) do
{:ok, token}
end
end
end
We’ll use libcluster to establish communication between our Elixir nodes. Here’s an example configuration that uses the Gossip strategy to form a cluster between nodes:
defmodule MessagePipeline.Application do
use Application
def start(_type, _args) do
topologies = [
gossip_example: [
strategy: Elixir.Cluster.Strategy.Gossip
]
]
children = [
{Cluster.Supervisor, [topologies, [name: MessagePipeline.ClusterSupervisor]]},
# Other children...
]
Supervisor.start_link(children, strategy: :one_for_one)
end
end
We’ll use Horde to manage our Broadway pipelines across the cluster. Horde ensures that each pipeline runs on exactly one node and handles redistribution when nodes fail.
Let’s add Horde’s supervisor and registry to the application’s supervision tree.
The UniformQuorumDistribution distribution strategy distributes processes using a hash mechanism among all reachable nodes. In the event of a network partition, it enforces a quorum and will shut down all processes on a node if it is split from the rest of the cluster: the unreachable node is drained and the pipelines can be resumed on the other cluster nodes.
defmodule MessagePipeline.Application do
use Application
def start(_type, _args) do
children = [
{Horde.Registry, [
name: MessagePipeline.PipelineRegistry,
members: :auto,
keys: :unique
]},
{Horde.DynamicSupervisor, [
name: MessagePipeline.PipelineSupervisor,
members: :auto,
strategy: :one_for_one,
distribution_strategy: Horde.UniformQuorumDistribution
]}
# Other children...
]
Supervisor.start_link(children, strategy: :one_for_one)
end
end
Each pipeline uses Broadway to consume messages from RabbitMQ and publish them to Google PubSub.
A strict, per-queue ordering is guaranteed by setting a concurrency of 1.
defmodule MessagePipeline.Pipeline do
use Broadway
alias Broadway.Message
def child_spec(opts) do
queue_name = Keyword.fetch!(opts, :queue_name)
pipeline_name = pipeline_name(queue_name)
%{
id: pipeline_name,
start: {__MODULE__, :start_link, opts}
}
end
def start_link(opts) do
queue_name = Keyword.fetch!(opts, :queue_name)
pipeline_name = pipeline_name(queue_name)
pipeline_opts = [
name: {:via, Horde.Registry, {MessagePipeline.PipelineRegistry, pipeline_name}},
producer: [
module: {
BroadwayRabbitMQ.Producer,
queue: queue_name,
connection: [
host: System.fetch_env!("RABBITMQ_HOST"),
port: String.to_integer(System.fetch_env!("RABBITMQ_PORT")),
username: System.fetch_env!("RABBITMQ_USER"),
password: System.fetch_env!("RABBITMQ_PASSWORD")
]
},
concurrency: 1
],
processors: [
default: [
concurrency: 1
]
],
batchers: [
default: [
batch_size: 100,
batch_timeout: 200,
concurrency: 1
]
]
]
case Broadway.start_link(__MODULE__, pipeline_opts) do
{:ok, pid} ->
{:ok, pid}
{:error, {:already_started, _pid}} ->
:ignore
end
end
def pipeline_name(queue_name) do
String.to_atom("pipeline_#{queue_name}")
end
@impl true
def handle_message(_, message, _) do
message
|> Message.update_data(&process_data/1)
end
@impl true
def handle_batch(_, messages, _, _) do
case publish_to_pubsub(messages) do
{:ok, _message_ids} -> messages
{:error, reason} ->
# Mark messages as failed
Enum.map(messages, &Message.failed(&1, reason))
end
end
defp process_data(data) do
# Transform message data as needed
data
end
defp publish_to_pubsub(messages) do
MessagePipeline.GooglePubsub.publish_messages(messages)
end
end
Finally, we need a process to monitor RabbitMQ queues and ensure pipelines are running for each one.
The Pipeline Manager periodically queries RabbitMQ for existing queues. If a new queue appears, it starts a Broadway pipeline only if one does not already exist in the cluster. If a queue is removed, the corresponding pipeline is shut down.
defmodule MessagePipeline.PipelineManager do
use GenServer
@timeout :timer.minutes(1)
def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
end
def init(_opts) do
state = %{managed_queues: MapSet.new()}
{:ok, state, {:continue, :start}}
end
def handle_continue(:start, state) do
state = manage_queues(state)
{:noreply, state, @timeout}
end
def handle_info(:timeout, state) do
state = manage_queues(state)
{:noreply, state, @timeout}
end
def manage_queues(state) do
{:ok, new_queues} = discover_queues()
current_queues = state.managed_queues
queues_to_add = MapSet.difference(new_queues, current_queues)
queues_to_remove = MapSet.difference(current_queues, new_queues)
Enum.each(queues_to_add, &start_pipeline/1)
Enum.each(queues_to_remove, &stop_pipeline/1)
%{state | managed_queues: new_queues}
end
defp discover_queues do
{:ok, conn} =
AMQP.Connection.open(
host: System.fetch_env!("RABBITMQ_HOST"),
port: String.to_integer(System.fetch_env!("RABBITMQ_PORT")),
username: System.fetch_env!("RABBITMQ_USER"),
password: System.fetch_env!("RABBITMQ_PASSWORD")
)
{:ok, chan} = AMQP.Channel.open(conn)
{:ok, queues} = AMQP.Queue.list(chan)
# Filter out system queues
queues
|> Enum.reject(fn %{name: name} ->
String.starts_with?(name, "amq.") or
String.starts_with?(name, "rabbit")
end)
|> Enum.map(& &1.name)
|> MapSet.new()
end
defp start_pipeline(queue_name) do
pipeline_name = MessagePipeline.Pipeline.pipeline_name(queue_name)
case Horde.Registry.lookup(MessagePipeline.PipelineRegistry, pipeline_name) do
[{pid, _}] ->
{:error, :already_started}
[] ->
Horde.DynamicSupervisor.start_child(
MessagePipeline.PipelineSupervisor,
{MessagePipeline.Pipeline, queue_name: queue_name}
)
end
end
defp stop_pipeline(queue_name) do
pipeline_name = MessagePipeline.Pipeline.pipeline_name(queue_name)
case Horde.Registry.lookup(MessagePipeline.PipelineRegistry, pipeline_name) do
[{pid, _}] ->
Horde.DynamicSupervisor.terminate_child(MessagePipeline.PipelineSupervisor, pid)
[] ->
{:error, :not_found}
end
end
end
Let’s not forget to also add the pipeline manager to the application’s supervision tree.
defmodule MessagePipeline.Application do
use Application
def start(_type, _args) do
children = [
{MessagePipeline.PipelineManager, []}
# Other children...
]
Supervisor.start_link(children, strategy: :one_for_one)
end
end
We should now have a working and reliable system. To quickly test it out, we can configure a local RabbitMQ broker, a Google Cloud PubSub topic, and finally a couple of Elixir nodes to verify that distributed pipelines are effectively run to forward messages between RabbitMQ queues and PubSub.
Let’s start by running RabbitMQ with the management plugin. RabbitMQ will listen for connections on the 5672 port, while also exposing the management interface at http://localhost:15672. The default credentials are guest/guest.
docker run -d --name rabbitmq \
-p 5672:5672 \
-p 15672:15672 \
rabbitmq:3-management
Next, install and use the gcloud CLI to create a Google Cloud project, a PubSub topic, and a a service account to access PubSub programmatically.
# Login to Google Cloud
gcloud auth login
# Create a new project (or use an existing one)
gcloud projects create message-pipeline-test
gcloud config set project message-pipeline-test
# Enable PubSub API
gcloud services enable pubsub.googleapis.com
# Create a topic
gcloud pubsub topics create test-topic
# Create service account for local testing
gcloud iam service-accounts create local-test-sa
# Generate and download credentials
gcloud iam service-accounts keys create ./google-credentials.json \
--iam-account local-test-sa@message-pipeline-test.iam.gserviceaccount.com
# Grant publish permissions
gcloud pubsub topics add-iam-policy-binding test-topic \
--member="serviceAccount:local-test-sa@message-pipeline-test.iam.gserviceaccount.com" \
--role="roles/pubsub.publisher"
It’s now time to start two terminal sessions where we can export the needed environment variables before running two Elixir nodes.
# In terminal 1
export RABBITMQ_HOST=localhost
export RABBITMQ_PORT=5672
export RABBITMQ_USER=guest
export RABBITMQ_PASSWORD=guest
export GOOGLE_PUBSUB_PROJECT_ID=message-pipeline-test
export GOOGLE_PUBSUB_TOPIC=test-topic
export GOOGLE_APPLICATION_CREDENTIALS_PATH=$(pwd)/google-credentials.json
export RELEASE_COOKIE=my-secret-cookie
iex --sname node1 -S mix
# In terminal 2 (same variables)
export RABBITMQ_HOST=localhost
export RABBITMQ_PORT=5672
export RABBITMQ_USER=guest
export RABBITMQ_PASSWORD=guest
export GOOGLE_PUBSUB_PROJECT_ID=message-pipeline-test
export GOOGLE_PUBSUB_TOPIC=test-topic
export GOOGLE_APPLICATION_CREDENTIALS_PATH=$(pwd)/google-credentials.json
export RELEASE_COOKIE=my-secret-cookie
iex --sname node2 -S mix
To verify cluster formation, from one of the nodes:
# List all nodes in the cluster, should show the other node
Node.list()
# Check Horde supervisor distribution
:sys.get_state(MessagePipeline.PipelineSupervisor)
Now let’s create some test queues on RabbitMQ and start publishing some messages.
# Download rabbitmqadmin if not already available
wget http://localhost:15672/cli/rabbitmqadmin
chmod +x rabbitmqadmin
# Create queues
./rabbitmqadmin declare queue name=test-queue-1
./rabbitmqadmin declare queue name=test-queue-2
# Publish test messages
./rabbitmqadmin publish routing_key=test-queue-1 payload="Message 1 for queue 1"
./rabbitmqadmin publish routing_key=test-queue-1 payload="Message 2 for queue 1"
./rabbitmqadmin publish routing_key=test-queue-2 payload="Message 1 for queue 2"
# List queues and their message counts
./rabbitmqadmin list queues name messages_ready messages_unacknowledged
# Get messages (without consuming them)
./rabbitmqadmin get queue=test-queue-1 count=5 ackmode=reject_requeue_true
One can also use the RabbitMQ management interface at http://localhost:15672, authenticate with the guest/guest default credentials, go to the “Queues” tab, click “Add a new queue”, and create “test-queue-1” and “test-queue-2”.
After a minute, the Elixir nodes should automatically start some pipelines corresponding to the RabbitMQ queues.
# List all registered pipelines
Horde.Registry.select(MessagePipeline.PipelineRegistry, [{{:"$1", :"$2", :"$3"}, [], [:"$2"]}])
# Check specific pipeline
pipeline_name = :"pipeline_test-queue-1"
Horde.Registry.lookup(MessagePipeline.PipelineRegistry, pipeline_name)
Now, if we publish messages on the RabbitMQ queues, we should see them appear on the PubSub topic.
We can verify it from Google Cloud Console, or by creating a subscription, publishing some messages on RabbitMQ, and then pulling messages from the PubSub subscription.
gcloud pubsub subscriptions create test-sub --topic test-topic
# ...Publish messages on RabbitMQ queues...
gcloud pubsub subscriptions pull test-sub --auto-ack
If we stop one of the Elixir nodes (Ctrl+C twice in its IEx session) to simulate a failure, the pipelines should be redistributed in the remaining node:
# Check updated node list
Node.list()
# Check pipeline distribution
Horde.Registry.select(MessagePipeline.PipelineRegistry, [{{:"$1", :"$2", :"$3"}, [], [:"$2"]}])
With our current implementation, pipelines are automatically redistributed when a node fail but they are not redistributed when a new node joins the cluster.
Fortunately, Horde supports precisely this functionality from v0.8+, and we don’t have to manually stop and re-start our pipelines to have them landing on other nodes.
All we need to do is enable the option process_distribution: :active on Horde’s supervisor to automatically rebalance processes on node joining / leaving. The option runs each child spec through the choose_node/2 function of the preferred distribution strategy, detects which processes should be running on other nodes considering the new cluster configuration, and specifically restarts those particular processes such that they run on the correct node.
defmodule MessagePipeline.Application do
use Application
def start(_type, _args) do
children = [
{Horde.Registry, [
name: MessagePipeline.PipelineRegistry,
members: :auto,
keys: :unique
]},
{Horde.DynamicSupervisor, [
name: MessagePipeline.PipelineSupervisor,
members: :auto,
strategy: :one_for_one,
distribution_strategy: Horde.UniformQuorumDistribution,
process_redistribution: :active
]}
# Other children...
]
Supervisor.start_link(children, strategy: :one_for_one)
end
end
This architecture provides a robust solution for processing ordered message streams at scale. The combination of Elixir’s distributed capabilities, Broadway’s message processing features, and careful coordination across nodes enables us to build a system that can handle high throughput while maintaining message ordering guarantees.
To extend this solution for your specific needs, consider these enhancements:
TSDgeos
@zzag:kde.org
