DEV Community: Andrea

RTO, Return To On-premise

Andrea — Sun, 21 Dec 2025 19:42:39 +0000

How we got there

The advantage of moving to the cloud was always clear: no big CAPEX, just a monthly fee and someone else will provide the compute, the network and everything else. As CTOs began to propose “lift and shift” strategies with an almost Pavlovian response, cloud adoption became the default setting, turning into a buzzword that eventually numbed us. We forgot that, at the end of the day, we were simply running our businesses on someone else's computer.

Agency & Fate Sharing

This numbness is usually interrupted when the “shared” nature of that computer becomes painfully obvious. In 2017, a single bug in a Cloudflare HTML parser leaked the private memory of thousands of unrelated companies into public search engine caches. But if that was a warning regarding privacy, the recent outage was a definitive lesson on availability. When a firewall adjustment intended to mitigate a vulnerability in the React framework was deployed, half of the internet went down because of another parser bug. Why should my shop go down because someone else is using React? There is nothing that any technical leader could have done to prevent that. And don't get me started about the guarantee that your customer data is going to be leaked (but you can blame someone else, right?)

What Changed?

In a centralized model, you trade the hassle of hardware for the risk of a blast radius that you cannot control. Luckily, the landscape has fundamentally changed since the early days of cloud migration. Today, technologies like Kubernetes, Terraform and declarative provisioning principles allow us to treat physical racks exactly like cloud regions. We are no longer managing servers; we are managing manifests. This shift allows us to pack a fully automated, self-healing cloud cluster into an on-premise environment that remains virtually untouched by human hands, controlled entirely by the same gitops workflows used by the big guys.

We finally have the technical maturity to reject the awkwardness of opening our digital shops in the house of our biggest competitor, as is the case with AWS and e-commerce.

RTO (Return to On-premise) is a little bit about nostalgia for server racks, but mostly it is about the ability to claim the right to go down because I fucked up.

Software Engineering for Xenophorms

Andrea — Tue, 24 Jun 2025 07:26:38 +0000

This is not a guide on how to build xenophorm software, it's just a description of how they do it.

Every solar cycle this creature awakens and positions itself before a foldable object made of very dense artificial material. It consists of two solid parts joined by a metallic hinge mechanism that allows expansion and collapse. When expanded, the upper section displays pulsating illumination with continuously shifting patterns, while the lower section is covered with small square buttons arranged in orderly rows.

The being presses these buttons in seemingly random sequences, causing alterations in the luminous patterns above. Its appendage movements are swift and precise - it appears to intuitively know which button to press to achieve specific photon emission configurations. This process continues for many consecutive hours.

Periodically the being ingests a dark liquid from a cylindrical vessel, after which its movements become more frantic and the luminous patterns shift more rapidly. The substance appears to have pseudoscientific properties that enhance the ritual's intensity.

The medicine induces loud vocalization such as "compiling," "git push," and "deployment failed" - apparently formulas necessary for proper process execution.

The surrounding environment contains numerous other rectangular devices, serpentine colored cables, and objects that emit constant humming sounds - likely auxiliary tools for this primitive software creation ritual.

Well, primitive by our standards, yet surprisingly effective for their level of technological development. They would probably be surprised to discover our much more efficient, yet basic, chorus-driven software engineering, where 300 individuals collectively sing code to a lazy snail, modifying its swaying patterns such that surfaces are imbued with binary saliva trails which, licked by herbivores, cause venom-induced paranoia, leading to Darwinian evolution towards engineering creatures capable of writing the chanted software.

Tilt: Improving Local Development with Production-Like Environments

Andrea — Sun, 06 Apr 2025 22:48:11 +0000

After reading about Tilt in "Optimizing Cloud Native Java" by Benjamin Evans, I decided to try it out. Tilt promises to reproduce production stacks for local development with hot reloading capabilities.

The Problem Tilt Solves

Local development environments are often broken because they're not part of CI. Despite attempts to include local development setup in CI, it's difficult to maintain. This leads to waiting for CI pipelines to finish or cobbling together commands and port-forwards to test services locally.

As a developer who relies on tight feedback loops for TDD, having a reliable local environment is crucial for productivity.

Implementation Example

I tried Tilt with a Node.js project that remotely controls my doorbell. Here's the Tiltfile I created:

load('ext://secret', 'secret_create_generic')

# Build Docker image from local directory
docker_build('ghcr.io/gurghet/domobot', '.')

# Deploy Kubernetes resources
k8s_yaml('infra/namespace.yaml')
k8s_yaml('infra/deployment.yaml')

# Deploy Mosquitto MQTT broker for local development
k8s_yaml('infra/local/mosquitto.yaml')

# Create Kubernetes secret from local environment variables
secret_create_generic(
    name='domobot-secrets', 
    namespace='domobot',
    from_env_file='.env.local'
)

# Configure Mosquitto MQTT broker resource and port forwarding
k8s_resource(
    'mosquitto',
    port_forwards=["1883:1883"]
)

# Configure the domobot deployment resource
k8s_resource('domobot', port_forwards=[])

With this single configuration file, I was able to launch a dockerized Node.js app alongside an MQTT broker service with the necessary secrets.

The Key Advantage: Instant Updates

The most impressive feature was the hot reload capability. When I fixed a bug in my application code and simply saved the file, Tilt automatically rebuilt the Docker image and updated the Kubernetes deployment instantly.

The UI clearly showed all components of the deployment, with warnings for any issues. Being able to see errors and fix them without manual rebuilding or redeploying dramatically improved my workflow.

Conclusion

Tilt effectively bridges the gap between local development and production environments. While implementing it for complex microservice architectures will require initial setup effort, the benefits of a reliable, self-healing local environment outweigh the investment.

The tool provides what Docker Compose alone cannot: truly seamless hot reloading of containerized applications in a Kubernetes context, creating a much tighter feedback loop for development.

Webhooks: A New Approach to an Old Problem

Andrea — Fri, 30 Aug 2024 07:25:12 +0000

As developers, we've all been there: knee-deep in a project, trying to integrate real-time data from a third-party service. Webhooks seem like the perfect solution... until they're not.

I once lost 800 euros because of a webhook. Sounds ridiculous, right? But it happened, and it taught me a valuable lesson about the pitfalls of real-time data integration.

I was working on a project to integrate Airbnb bookings with a property management system. The damn system was not idempotent on reads! So I could not use the API to read the latest booking because it will disappear the next time. So I choose to use webhooks.

A small error in the webhook documentation—a field that sometime was missing—meant that booking notifications weren't coming through because of a parsing error. We missed a reservation, and a guest showed up to a double-booked apartment. Sorting out alternative accommodation at the last minute cost us 800 euros and a fair bit of stress.

It wasn't just about the money. It was the realization that our entire system was vulnerable to such a simple mistake. One missed webhook, and everything falls apart.

This experience got me thinking: there has to be a better way to handle real-time data without the constant anxiety of missing a beat. And that's exactly what I want to share with you today.

Webhook-to-API

Enter a new service that's changing the game. It's simple: you get a unique URL like this:

https://webhook.passaglia.it/<your-prefix>/webhook

Point your webhook-happy services here, and the magic begins. All that real-time data gets collected and stored. The best part? You can access it anytime with a simple API call:

https://webhook.passaglia.it/<your-prefix>/api/data

It's like having a personal assistant for your webhooks. They collect all the messages, organize them, and give you a neat summary whenever you ask.

Think about the possibilities:

Managing team projects on OneDrive? No more sifting through individual file upload notifications. One API call gives you a list of all uploads in the last 24 hours.
Tracking GitHub commits for your team? Forget about processing each webhook in real-time. Collect the data and analyze it on your terms. "Most active contributors this week" report? Easy.
Running an online store? Instead of reacting to every single order webhook, aggregate that data. Analyze order patterns, identify peak hours, track product popularity - all with simple API calls.
Working on an IoT project? Let this service be your data aggregator. All those sensor readings collected in one place, ready for analysis when you need it.

This isn't about replacing webhooks. It's about giving us, the developers, more control. It's about saying, "Yes, I want that real-time data, but I'll process it when I'm good and ready."

In my years of coding, I've learned that flexibility is key. This service offers just that - the immediacy of webhooks with the control of traditional APIs. It's a safety net that could have saved me 800 euros and a lot of headaches.

So the next time you're starting a project and webhooks come up, remember: you have options. You don't have to be at the mercy of real-time data. Take control, simplify your architecture, and focus on what really matters - building great software.

After all, isn't that why we got into this game in the first place?

I Accidentally The Whole Cloud... And It Didn't Matter

Andrea — Sun, 16 Jun 2024 10:36:56 +0000

Want to build a Wordpress from scratch in 30 seconds? Ok, start the timer now!

So, I wanted my Wordpress website, but, being a software engineer, I didn't want to pay for it, not anything except the little more than the electricity needed to run a machine. Also, in the past I accidentally the whole website (more than once 🥲). To be clear, I did have the backup, but was too lazy to set up the whole system from scratch. Create the server, spin up the database, run the migration, and all that jazz.
I don't want this to happen ever again 😡!
So I set to myself the rule that whatever I build, even the most innocent Telegram bot to randomly choose the pizza for me (by the way why it doesn't have AI yet to understand my tastes?)

Whatever I build from now on will be IaC, so that if I accidentally the whole cloud account again, I would be able to recreate it with the push of a button (preferably Enter, after issuing the terraform plan command).

The machines

Hetzner is a thrifty hosting, I don't know what that means, let me check the dictionary... oh makes sense. Well you can have the cheapest server for just a few euros a month, I always choose ARM because it seems cooler (figuratively and literally). But Intel has the same prices more or less.

This is the bulk of the expense for our new Wordpress website. Good.

From the hetzner cloud get your read/write API token.

The machines, as code

We’re going to use Terraform to automate everything. Terraform is an open-source IaC tool that lets you define and provision your infrastructure with code. No more manual setup or hoping you didn’t screw up a step. With Terraform, you get consistency, fewer errors, and version control. Plus, if you need to wipe everything and start fresh, it’s just a couple of commands away. It’s the perfect tool to make sure your infrastructure is always up and running exactly how you want it.

We are going to use some free cloud services, but none of them are necessary, we'll use them for their power to streamline workflow:
I would highly suggest to get a Terraform Cloud account, the free plan can manage 500 resources which is plenty.
Get an account at Doppler.com, it's basically a vault and will be useful to let the secret survive a total wipeout event. It also has a free tier. Get the API token for that as well (read only is fine for now).

Once you have these two create a new folder for this project and make it a git repository. For simplicity I'm going to assume we have a GitHub repo called wordpress-in-a-jar (in theory could be GitLab, BitBucket or even hosted on-premise). Yes, get an API token for that too, we'll need that to read/write the repository content.

Create a main.tf with the following:

terraform {
  required_providers {
    hcloud = {
      source  = "hetznercloud/hcloud"
      version = "~> 1.0"
    }
  }
}

provider "hcloud" {
  token = var.hcloud_token
}

variable "hcloud_token" {
  type = string
}

variable "github_token" {
  type = string
}

resource "hcloud_server" "master_node" {
  name        = "geppetto"
  image       = "ubuntu-20.04"
  server_type = "cax11"
  location    = "fsn1"
  user_data   = templatefile("${path.module}/cloud-init.tpl", {
    github_token  = var.github_token
  })

  ssh_keys = [data.hcloud_ssh_key.gaia_key.id]
}

output "ip_address" {
  value = hcloud_server.master_node.ipv4_address
}

resource "hcloud_ssh_key" "gaia_key" {
  name        = "Public key of my MacBook Pro (Gaia)"
  public_key  = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQDUAl/qrUV1Nkcd7fPbYAahOAg7p4Nn5+Gkv5Y1lQ/Hm7DqSJki9mmxhtuB/HHV3pZuriAzVJKpee8q8p55EnWhv9xw04oHBXYuJYzkU0kNMiZGMgh/Z8BNkY7QBqitDLOeCNk8gKpKYY0kbINvUaWUNy/JQdmLUu9erCzbkkC0k3KLTlVRr6ZyKuJ6yHX9zYHDJRw9iO+SKA7V/fFVBZtxfYXNN0GaDw6+33z7A7pxbt4wlCuFir2AYTUcU6E2jwrtpq9gwJ0dXiiOW5H/RRGJ1D3VDIcag+Zy7p54K3fH2KOgjujbPq6SS8zJ8/GE+iHCCVxhLnXLin66rRUOIbYVzxPtryX+f4fAxfxTWKLMNWWIVFa11/FOJ792j9MIuYvV/dn3nICsBSQToGQ94A7LoN6W0j4INViHbkzEZVaXQth2urFQ/1NmJGnQkbRR8/XU4ej06WUtie9oGjNY2SXOKVahjBWoybunbtKuv4gtz/XEQKjQDU4T1qXJ6k11jEM= gurghet@Gaia.local"
}

That's my actual public key of my laptop, you have to put your own key there, otherwise I'll be able to access your machine :D

Cloud-init

The line referring to the cloud-init.tpl is calling built-in facility (in most server operating systems) for initializing and configuring cloud instances. It's very complicated how it works, has many modules and functionalities but we'll only use a few.

#cloud-config
runcmd:
  - export GITHUB_TOKEN=${github_token}
  - export HOME=/root
  - export XDG_CONFIG_HOME=/root/.config
  - curl -sfL https://get.k3s.io | sh - # install k8s
  - export KUBECONFIG=/etc/rancher/k3s/k3s.yaml
  - until nc -z localhost 6443; do sleep 1; done # wait until k8s is live
  - curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash # get helm
  - curl -s https://fluxcd.io/install.sh | sudo bash
  - kubectl apply -f https://github.com/fluxcd/flux2/releases/latest/download/install.yaml --validate=false
  - flux bootstrap github --owner=gurghet --repository=wordpress-in-a-jar --branch=master --path=./flux/manifests

In the last command we are provisioning the node with flux CD, which will watch every our step in the git repository and reconciliate until the cluster looks exactly like our code.

Similarly here, you should use your github handle instead of gurghet and possibly other parameters accordingly. They should all correspond to a real GitHub repository that you created and uploaded, with the above two files inside basically.

Doing the stuff

Now that your repository is accessible on GitHub, you should actually tell terraform to create all this on Hetzner. You should use the following commands:

terraform init
terraform apply -var 'hcloud_token=<your hetzner token>' -var 'github_token=<your github token>'

The firt command prepares the terraform environment and initializes the state, the second actually executes the code we wrote. We are manually passing the secrets now, but we will move them away later. We want a push-button experience, no manual secret management required.

The plan will spit out the ip address of your new master node Geppetto, you should then be able to log in with ssh:

ssh root@<Geppetto's ip>

Once inside you can check that the provisioning was succcessful by tailing the logs, I got:

root@geppetto:~# tail -f /var/log/cloud-init-output.log
✔ GitRepository reconciled successfully
◎ waiting for Kustomization "flux-system/flux-system" to be reconciled
✔ Kustomization reconciled successfully
► confirming components are healthy
✔ helm-controller: deployment ready
✔ kustomize-controller: deployment ready
✔ notification-controller: deployment ready
✔ source-controller: deployment ready
✔ all components are healthy
Cloud-init v. 24.1.3-0ubuntu1~20.04.1 finished at Sat, 15 Jun 2024 21:24:35 +0000. Datasource DataSourceHetzner.  Up 113.81 seconds

If this worked you should find a new commit to your flux. This initial commit contains all the necessary files for Flux to manage your cluster state.

Congratulations! You have a master node waiting for your commands. But let's say you want to call it a day and continue building this tomorrow. You don't need to keep your machine running and spending all your money. You can use terraform destroy and then, tomorrow after a coffee, terraform apply like above to instantly recreate your cluster.

Spinning up the wordpress

Everything you write in the manifests folder will become true. So let's create a new folder for our Wordpress website: we'll call it "bot-buffet" an all-you-can-click Telegram bot marketplace. We'll use the excellent bitnami helm. This is how the directory structure is looking now:

We are going to need 4 charts here:

Bitnami helm repository
A namespace for bot-buffet
The HelmRelease itself with Wordpress
An Ingress to instruct Traefik to route to it

I said charts but let's actually but everything inside wordpress-helm.yaml because doing like this increase coesion and makes it easier to read the code.

The first 2 are straightforward:

---
apiVersion: source.toolkit.fluxcd.io/v1
kind: HelmRepository
metadata:
  name: bitnami
  namespace: flux-system
spec:
  interval: 10m
  url: https://charts.bitnami.com/bitnami
---
apiVersion: v1
kind: Namespace
metadata:
  name: bot-buffet
---

The bitnami chart needs some value, in particular we can choose name and email, but importantly we must disable the built-in loadbalancer (we don't want to spend money, remember?) and let's instead use Traefik.

---
apiVersion: helm.toolkit.fluxcd.io/v2
kind: HelmRelease
metadata:
  name: bot-buffet-wordpress
  namespace: bot-buffet
spec:
  interval: 5m
  chart:
    spec:
      chart: wordpress
      version: "22.4.5"
      sourceRef:
        kind: HelmRepository
        name: bitnami
        namespace: flux-system
  values:
    wordpressUsername: "gurghet"
    wordpressEmail: "gurghet@proton.me"
    wordpressBlogName: "Bot Buffet"
    service:
      type: ClusterIP
    ingress:
      enabled: true
      ingressClassName: "traefik"
---

Perfect! The last think will be the Ingress entry to tell Traefik to actually send in all the customers that come from port 80.

---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: bot-buffet-wordpress-ingress
  namespace: bot-buffet
  annotations:
    traefik.ingress.kubernetes.io/router.entrypoints: web
spec:
  rules:
  - http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: bot-buffet-wordpress
            port:
              name: http
---

The service bot-buffet-wordpress is hardcoded because this is what the helm chart above will create.

At this point you just need to commit and push to github this new file. The flux will pick it up and magically create your Wordpress! Try it by pointing your browser to the ip of the machine.

Of course there is a bunch of things that we still need to do: where is my domain with TLS certificate? Where are all the password to access my Wordpress and Database? How do I debug if something goes wrong? Where are my Telegram bots? I'm going to tell you next time.

Did you check the timer? Are we over 30 seconds? Oh crap, ok let's start from scratch. Do the terraform destroy. Ok, Ready?

terraform apply

Done!

References: https://github.com/gurghet/wordpress-in-a-jar

Bidirectional topic mirroring

Andrea — Sun, 25 Feb 2024 23:03:44 +0000

The tool of choice when it comes to disaster recovery in Kafka, is Mirror Maker. It's often configured to keep a stand-by cluster full of the latest data to take over just in case something happens to the main cluster.

The same tool is also used for geo-replication. In this case some selected topics are sent to another region to have consistency of data between the clusters and deliver an updated state.

This last setup however only works in one direction normally and, while Mirror Maker does support bidirectional replication, it does so in an asymmetric way by appending the name of the originating cluster to the topic, this is done to avoid the infinite mirror effect of a message being replicated back and forth. Consumers and producers have to be aware of this asymmetry and the topology has to be more complex than necessary.

This spurred in a few cases where the setups used custom code to manage aggregating the events. Complexity is still lurking because of the inherent manual setup of Mirror Maker to manage these bidirectional flows effectively. The main challenge is the duplication and potential loss of messages due to network issues or misconfigurations. These complexities are not only operational but also introduce a high risk of data inconsistency between clusters.

An alternative approach to Mirror Maker for bidirectional topic mirroring in Kafka is to use more advanced replication tools or services that are designed with bidirectionality in mind from the start. Tools like Confluent Replicator offer enhanced features over Mirror Maker, including improved fault tolerance, easier configuration, and better support for bidirectional replication. However, these are usually part of a cloud offering and have a price tag associated.

Today I tried some alternatives with Apache Pulsar and have prepared a quick PoC you can try yourself.

Setting Up Bidirectional Mirroring with Pulsar

Quick Guide: Follow these streamlined steps to configure bidirectional topic mirroring using Apache Pulsar.

Following the setup outlined in the README from the pulsar-geo-replication GitHub project linked above, let's walk through the commands necessary to set up bidirectional topic mirroring using Apache Pulsar. This guide assumes you have already cloned the project and have Docker and Docker-compose installed on your system.

Start the Pulsar Clusters

Navigate to the project directory where the docker-compose.yml file is located. Initiate the clusters by running:

docker-compose up

This command starts two Pulsar clusters, named cluster-a and cluster-b, running in Docker containers.

Connect the Clusters

After the clusters are up, you need to establish a connection between them. Execute the following commands to link cluster-a to cluster-b and vice versa:

./pulsar-admin --admin-url http://localhost:8080 clusters create cluster-b --broker-url pulsar://broker-edge1:6650 --url http://broker-edge1:8080
./pulsar-admin --admin-url http://localhost:8081 clusters create cluster-a --broker-url pulsar://broker:6650 --url http://broker:8080

These commands configure each cluster to recognize the other, enabling them to replicate data between each other.

Create Tenants and Namespaces

Next, set up tenants and namespaces in both clusters to facilitate the replication:

./pulsar-admin --admin-url http://localhost:8080 tenants create edge1 --allowed-clusters cluster-a,cluster-b
./pulsar-admin --admin-url http://localhost:8081 tenants create edge1 --allowed-clusters cluster-a,cluster-b
./pulsar-admin --admin-url http://localhost:8080 namespaces create edge1/replicated --clusters cluster-a,cluster-b
./pulsar-admin --admin-url http://localhost:8081 namespaces create edge1/replicated --clusters cluster-a,cluster-b

This configuration ensures that any topic created in the edge1/replicated namespace will automatically be replicated across both clusters.

Topic Creation

Now, let's create a topic in the replicated namespace:

./pulsar-admin --admin-url http://localhost:8080 topics create persistent://edge1/replicated/events

This topic will serve as the conduit for messages meant to be mirrored between the clusters.

Testing the Setup

Open a consumer for both cluster to check if the messages will come through:

./pulsar-client --url http://localhost:8080 --listener-name external consume --subscription-name "sub-a" persistent://edge1/replicated/events -n 0

The listener name is imporant because we can't access the Docker DNS from the host machine, so we configured in the docker-compose.yaml an external listener that is advertised from the broker to the pulsar client, which the pulsar client will then use to connect to the broker after the first connection.
Open a second terminal and start listening for messages on cluster b:

./pulsar-client --url http://localhost:8081 --listener-name external consume --subscription-name "sub-b" persistent://edge1/replicated/events -n 0

and in a third terminal you produce a message:

./pulsar-client --url http://localhost:8080 --listener-name external produce persistent://edge1/replicated/events --messages "Hello world produced to cluster a"

You will see the same message in both clusters.

Extra

It's also possible to avoid replicating on a per-message basis:

./pulsar-client --url http://localhost:8081 --listener-name external produce --disable-replication persistent://edge1/replicated/events --messages "Hello world produced to cluster b not replicated"

And you will see the message only on cluster b, while it will never appear on cluster a.

By following these steps, you've successfully set up bidirectional topic mirroring with Apache Pulsar. This method offers a more straightforward and reliable approach to data replication across clusters, ensuring high availability and consistency for distributed systems.

A closer look at the lifecycle of a send block in the Nano network

Andrea — Sun, 28 Mar 2021 19:58:54 +0000

How does a send block get memorized in the Nano network

A bystander look at the C++ reference implementation

I've spent some time looking at the Nano current reference implementation. The codebase is huge so it wasn't an easy task. I wanted to focus on a precise question: what is the lifecycle of a send block? These are my findings.

Conception

Since this piece will be about a send block, everything about creating a new chain is out of scope. Let's imagine a user wants to send some raws. My node will create a message with a header similar to this:

...
network: live
protocol version: 19
message type: publish
extensions:
  block type: send

I will pretend a random user with a balance of 700 raw wants to send 10 raw.
If we drill into the block information we'll find something like this:

previous: BBE55A35F79F887...
link/destination: 9A2726664A18FE5...
balance: 690
work: 14b3bc748f2c8e93
signature: B421B88AFBEDFC...

The balance is 690 raw because it was 700 and I'm sending 10 raw.
The node then will send this message to its peers.

Another node receive the message

For each peer there is an already established TCP connection and after a message is processed a new message listener is created.
This is how the listener is installed in bootstrap_server.cpp:151

void nano::bootstrap_server::receive ()
{
    // ...
    socket->async_read (receive_buffer, 8, [this_l](boost::system::error_code const & ec, size_t size_a) {
        // ...
        // Receive header
        this_l->receive_header_action (ec, size_a);
    });
}

Which will put whatever we receive through the TCP connection into the receive_buffer.
The function receive_header_action is immediately after and reads like this

void nano::bootstrap_server::receive_header_action (boost::system::error_code const & ec, size_t size_a)
{
    if (!ec)
    {
        // ...
        nano::bufferstream type_stream (receive_buffer->data (), size_a);
        auto error (false);
        nano::message_header header (error, type_stream);
        if (!error)
        {
            auto this_l (shared_from_this ());
            switch (header.type) {...}
        }
    }
    else
    {
         // error management ...
    }
}

What happens above is that the head of the receive_buffer is assigned to type_stream and type_stream is used to instanciate a message_header class. The logic in the constructor will deserialize the stream and, in particular, will fill the header.type attribute. This is because, provided no error happened, the next thing we do will depend on the header.type (the switch construct). Let's see the case for a publish message.

case nano::message_type::publish:
{
    socket->async_read (receive_buffer, header.payload_length_bytes (), [this_l, header](boost::system::error_code const & ec, size_t size_a) {
        this_l->receive_publish_action (ec, size_a, header);
    });
    break;
}

It's installing another listener, on the same buffer. The handler will call the receive_publish_action function in the same file, which validates the work in the carried block. It then adds the message to the requests deque. This will be ultimately processed by the request_response_visitor which in turn puts the message into the entries deque of the tcp_message_manager.

Processing message entries

At this point the network class enters the stage. When initialized, this class runs the process_messages loop at tcp.cpp:279.

void nano::transport::tcp_channels::process_messages ()
{
    while (!stopped) // while we are not shutting down the node
    {
        auto item (node.network.tcp_message_manager.get_message ());
        if (item.message != nullptr)
        {
            process_message (*item.message, item.endpoint, item.node_id, item.socket, item.type);
        }
    }
}

Internally the process_message, makes sure we have a channel open with the message originator. Then it creates a network_message_visitor relative to the channel and processes the publish message according to the following function in network.cpp:

void publish (nano::publish const & message_a) override
{
    // ... logging and monitoring logic ...
    if (!node.block_processor.full ())
    {
        node.process_active (message_a.block);
    }
    // ...
}

where process_active adds the block inside the message to both the block_arrival and the block_processor. The latter is responsible for putting the block into the block dequeue.

Block processing

Whenever a node class is instanciated it spawns a block processor thread. This thread has an infinite loop in blockprocessor.cpp inside the function process_blocks. This starts a transaction that, after acquiring various locks, processes a batch of blocks. The processing of a single block is defined in the process_one function and relies on a ledger_processor defined in ledger.cpp, at least for the send block we're interested in.

The full logic can be found in ledger.cpp in the send_block function. At its core it's a pyramid of ifs which try to account for all possible things that might go wrong. For example if the work of of the block is sufficient (note that we already checked this when we received the block from another node).

At the top of the pyramid we finally execute the instruction

ledger.store.block_put (transaction, hash, block_a);

which physically adds the block to the permanent storage.

Conclusion

This is not the end of the life of this block. In fact it would terminate when the block is cemented. Cementing is a different process that involves consensus, thus the block could be even be deleted if, for example was detected as a double spend. I'll write about this in another article.

Composing complex programs with ZIO

Andrea — Mon, 01 Mar 2021 22:50:29 +0000

There is a fundamental mental shift a procedural programmer must make when embracing functional programming. In this lesson we'll learn to visualize complex ZIO programs to compose them correctly.

The art of type chasing

Types in Scala give us confidence that the program is doing what we want and that the quality of the data is high enough. With ZIO they also help us catching nasty bugs at compile time instead of runtime.

The canonical ZIO program has this type:

ZIO[ZEnv, Nothing, ExitCode]

This is our end game, every ZIO program we can possibly conceive, if we transform it into (chase) this type, then we can easily run it.

Usually what we start with will be something like

ZIO[ZEnv with MyDatabase, DomainSpecificErrors, Unit]

to make it coincide with our canonical ZIO, we need to act in two different directions.

The environment type needs to become more generic (up to ZEnv)
The error and return type need to become more specific (error down to Nothing and return to ExitCode)

By using ZIO provided methods we can creatively chase the canonical types.

Cat and mouse chasing

As fun example of type chasing we can create a program of a cat chasing a mouse.

We will have a mouse program and a cat program that will run concurrently. Both programs will share a variable to track the mouse distance from the cat.

Creating a variable

Our variable, tracking the lead the mouse has on the cat, needs to be shared by both the cat and the mouse program.

To create a variable that is safe to use concurrently we can write the following program

import zio.Ref

val makeMouseLead: UIO[Ref[Int]] = Ref.make(10)

This is a UIO program which expands to a ZIO[Any, Nothing, Ref[Int]]. It can run in any environment, never fails, and returns a Ref of an integer.

Ref is a reference to an immutable variable that we can check and swap. You can picture a Ref like a single place datum cradle. We will use it to read the distance between cat and mouse and update it whenever either of them advances.

Digital mouse that runs away

To represent the mouse getting away from the cat we can design a simple recursive program

import zio.clock.sleep
import zio.console.putStrLn
import zio.random.nextIntBounded
import zio.duration._
import zio._

def mouseProgram(mouseLead: Ref[Int]): URIO[ZEnv, Unit] =
  for {
    cm <- nextIntBounded(2)
    _  <- mouseLead.update(_ + cm)
    _  <- putStrLn(s"Mouse advances by $cm cm")
    _  <- sleep(100.milliseconds)
    _  <- mouseProgram(mouseLead)
  } yield ()

Let's dissect this automatic mouse:

We are accepting the reference to an Int variable, the mouse lead on the cat.
A random number, less than 2, is drawn, representing how many centimeters the mouse advances.
We add the above number to the current lead and put the sum in the Ref. This will be the new mouse lead.
Print on the screen the mouse progress
We then wait 100 milliseconds
Recursively call the mouse program

Notice that we are using functions coming from zio.clock, zio.console and zio.random, all contained in ZEnv so our final type is URIO[ZEnv, Unit].

I wrote ZEnv instead of Clock with Console with Random. This is allowed because I can always pick a more specific type as environment. The opposite is true for error type and return type.

This program is recursive, it will never return the Unit we promised. To make the return type more expressive it would have to return Nothing meaning it will never do. But, as said above we cannot put a more specific type for the return channel.

The problem is that the compiler didn't detect our infinite recursion and assumed at some point we are going to return Unit. There is a way we can explicitly say that the mouse will run forever.

def mouseProgram(mouseLead: Ref[Int]): URIO[ZEnv, Nothing] =
  for {
    cm    <- nextIntBounded(2)
    _     <- mouseLead.update(_ + cm)
    _     <- putStrLn(s"Mouse advances by $cm cm")
    _     <- sleep(100.milliseconds)
    recur <- mouseProgram(mouseLead)
  } yield recur

By making the return type the mouse program itself we can force write Nothing and the compiler will be happy.

Thanks to the ZIO library we can actually do this by skipping the recursive step entirely:

def mouseProgram(mouseLead: Ref[Int]): URIO[ZEnv, Nothing] =
  (for {
    cm    <- nextIntBounded(2)
    _     <- mouseLead.update(_ + cm)
    _     <- putStrLn(s"Mouse advances by $cm cm")
    _     <- sleep(100.milliseconds)
  } yield ()).forever

In this code, we deleted the recursive call and wrapped the for comprehension in parenthesis. With forever we created a new program that repeats the for comprehension forever.

Digital cat that chases

The third program we are going to write is the dual of the mouse. The cat is very similar, but it runs faster and, for every centimeter it advances, the mouse lead diminishes instead of increasing.

def catProgram(mouseLead: Ref[Int]): URIO[ZEnv, Unit] =
  for {
    cm   <- nextIntBounded(5)
    lead <- mouseLead.updateAndGet(_ - cm)
    _    <- putStrLn(s"Cat advances by $cm cm, mouse is $lead cm away")
    _    <- sleep(100.milliseconds)
    _    <- catchDetector(lead, mouseLead)
  } yield ()

The cat program will actually finish at some point: the last instruction uses a special program called catchDetector. It considers two possible scenarios: either the mouse has still some lead or the cat is at or past the mouse.

def catchDetector(currentLead: Int, mouseLead: Ref[Int]): ZIO[ZEnv, Nothing, Unit] =
  currentLead match {
    case lead if lead > 0  => catProgram(mouseLead)
    case lead if lead <= 0 => putStrLn("Cat catches the mouse!")
  }

Here, if the mouse is not reached yet, the cat program is called (recursively), otherwise the catchDetector returns with a program that prints "Cat catches the mouse!".

We can make the program even more fun by including a third possibility: if the cat goes past the mouse, instead of declaring victory, we assume the mouse escaped. We can express this as an exception. From the point of view of the cat program, this is certainly not a welcome outcome.

First, let's declare the exception as a case class.

case class MouseEscapedException(catLead: Int)

The variable catLead will tell us how much the cat went past the mouse.

Secondly, we need to add the third case in the catchDetector.

def catchDetector(currentLead: Int, mouseLead: Ref[Int]): ZIO[ZEnv, MouseEscapedException, Unit] =
  currentLead match {
    case 0                 => putStrLn("Cat catches the mouse!")
    case lead if lead > 0  => catProgram(mouseLead)
    case lead if lead < 0 => ZIO.fail(MouseEscapedException(lead))
  }

Now if the currentLead is strictly less than 0 we return a program that fails with MouseEscapedException. This makes our program fail in turn.

Since the cat program is calling this, we need to update the signature by adding the error modality:

def catProgram(mouseLead: Ref[Int]): ZIO[ZEnv, MouseEscapedException, Unit] = ...

Putting it all together

We have created a total of 4 ZIO programs. This is a good time to list them all.

UIO[Ref[Int]] will give us an integer variable
URIO[ZEnv, Nothing] programs the mouse to run
ZIO[ZEnv, MouseEscapedException, Unit] programs the cat to chase
ZIO[ZEnv, MouseEscapedException, Unit] subroutine of the cat that can either succeed with a Unit or trigger failure with a MouseEscapedException.

The diagram below is another representation of these programs.

Notice, at the center, our canonical ZIO program URIO[ZEnv, ExitCode]. In green I represented the environment channel. The canonical ZIO program will provide you a perfectly good ZEnv (filled green dot). The program that creates the mouse and also the cat, conversely, want the ZEnv (hence the hollow green dot).

In blue I represented the success channel. The mouse catch detector may return a unit when a capture happens, this gets transmitted upwards). The mouse program runs forever, and returns nothing, that's why it doesn't have any blue line going outwards. The canonical ZIO waits patiently for a ExitCode and won't accept much else.

In red we see the failures channel that, from the catch detectors propagate upwards. The cat, in turn, may propagate the error.

We have to run the cat and the mouse program, to do that we have to chase the canonical ZIO types.

Environment channel

There is not a lot to do for the green lines. ZEnv with ZEnv, everything matches. We won't worry about this.

Failure channel

The canonical ZIO doesn't want to deal with errors we have to stop them before they reach the boundaries of our program. One way to do this is to turn the error into a program that never fails (a ZIO with Nothing as error type); putStrLn is one such a program. We could log the error to screen and this would return a success Unit type. A message saying that the mouse escaped will be sufficient.

Success channel

The only type accepted here by the canonical ZIO is ExitCode. We said this before, but it's worth repeating that ExitCode is just a puffed up integer we give to the operating system. We can always return ExitCode.success, an alias for 0. This means we consider the cat both catching and missing the mouse a success.

The above three conditions are satisfied by compositeProgram.

val compositeProgram: URIO[zio.ZEnv, ExitCode] =
  for {
    lead <- makeMouseLead
    exitCode <- catProgram(lead)
      .raceFirst(mouseProgram(lead))
      .catchAll(exc => putStrLn(s"Cat was ${-exc.catLead} cm ahead when it lost the mouse."))
      .as(ExitCode.success)
  } yield exitCode

Check out the complete code.

Click "Run" a few times to see all possible outcomes.

Explanation of `compositeProgram`

Our idea to chase types is a very small for comprehension. In the first place we map over the makeMouseLead program to create the variable. Secondly we run both the mouse and the cat with .raceFirst which will concurrently spin them up and wait for the first to terminate either with a success or a failure. We know that the mouse program doesn't terminate so we are basically waiting the cat program. This however leaves us with a ZIO[ZEnv, MouseEscapedException, Unit].

To turn the error into a Nothing we use .catchAll and we pass an infallible program: putStrLn. To turn the Unit into an exit code we could simply .flatMap it to an ExitCode but when we are not interested in the result we can use .as which is a .flatMap that ignores the preceding result.

All these transformations are summarized in the below diagram which now looks complete.

Bonus section

But wait, there is more. Instead of using the .catchAll and .as(ExitCode.success) functions, ZIO gives us a nice shortcut that does something very similar and that will work in the majority of cases. You can just add .exitCode and this will either print an error (and exit with 1) or succeed (and exit with 0). Not precisely our behavior, we lose the nice error message and we exit with an error when the mouse escape, but good enough: our program is simpler and we save one line.

val compositeProgram: URIO[zio.ZEnv, ExitCode] =
    for {
      lead <- makeMouseLead
      exitCode <- catProgram(lead)
        .raceFirst(mouseProgram(lead))
        .exitCode
    } yield exitCode

Building a Bitcoin price ticker

Andrea — Sat, 09 Jan 2021 20:23:20 +0000

Learning functional programming

Advanced track

In this series I want to explore fundamental concepts and tools of functional programming by building small applications.

Streams

A stream is a conceptual device that abstracts the way in wich we access and process data. A stream at its core is just a sequence of data. However, differently from an array or a list, a stream doesn't necessarily have a beginning or an end.

In real life we have some example of streams. A ticker for the price of Bitcoin for example. It's a stream because it's clearly a timed sequence of data. It does have a beginning (when Bitcoin was created) but it never ends.

Building a Bitcoin price ticker

In this issue you'll build a BTC ticker that runs on your screen. This way you will always know when to buy and when to sell.

Design

A simple design you can draft in two seconds is the one sketched above. We'll chose some API, parse the output, turn it into a number and finally print it on our screen (or physical ticker if you are retro).

Programming

A very first draft of the program might be like this:

ZStream
  .repeat("100.000") // generate some fake data
  .take(3) // let's not overload
  .map(identity) // placeholder for parsing
  .map(priceString => BigDecimal(priceString))
  .foreach(price =>
    putStrLn(
      s"BTC-EUR: ${price.setScale(2, BigDecimal.RoundingMode.UP)}"
    )
  )

You can go ahead and launch this. In the first line we are getting a hold of the ZStream object, it contains the repeat method wich just infinitely produces whatever we pass: a string in this case. We limit it to 3 strings, otherwise it will just overload our machine and never stop. The parsing for now is impersonated by an identity function. We then leverage the BigDecimal parsing capabilities to have a number to work with. Lastly the foreach method is used, which is a terminal method. It will wrap up the stream by printing on the screen a well formatted price tick.

We could have even skipped the conversion to BigDecimal and back if we don't care about having control on rounding.

ZStream
  .repeat("100.000")
  .take(3)
  .map(identity)
  .foreach(price => putStrLn(s"BTC-EUR: $price"))

Now we go hunting for a good API. Like in the cooking TV show I’ve already done this, but you can look for yourself and I recommend ProgrammableWeb.

I’m going to use the Kraken public API which is very basic. It returns a json of this form:

{
    "error": [],
    "result": {
        "XXBTZEUR": {
            ... other stuff
            "c": "last trade price",
            ... other stuff
        }
    }

We are interested in the current price which is the last trade price. We could (and should) use a json parser to extract this. However that would open a can of worms, so I’m going to use a regular expression:

def parsePrice(json: String): String = {
  val regex = raw"""(?s).*"c":\["([0-9.]+).*""".r
  json match {
    case regex(price) => price
  }
}

So we have our parsing step, we finally need to actually call the API. Again, there are super-cool http clients we could use; instead for simplicity's sake we'll use the scala built-in client.

val tickerURLCall: Task[String] = IO(
  io.Source
    .fromURL("https://api.kraken.com/0/public/Ticker?pair=BTCEUR")
    .mkString
)

We can now put everything together in the program. We will use tickerURLCall instead of the fake data and the priceParce instead of the identity.

  val program: ZIO[Console with Clock, Throwable, Unit] = ZStream
    .repeatEffect(tickerURLCall)
    .throttleShape(1, 5.seconds)(_ => 1) // let's slow down
    .map(parsePrice)
    .foreach(price =>
      putStrLn(
        s"BTC-EUR: $price"
      )
    )

Node the addition of the throttleShape function. This is because we don't want to call the API as fast as we can, but we want a constant slow rate. The code says 1 call every 5 seconds. The function we supply in the second pair of brackets is a function to compute the weight of the result (to shape the throttle accordingly) we consider all responses equal to 1.

This last feature is useful if you have pagination and you don't know how many items you have for each page. You could create a stream that outputs 10 items per minute and sometimes the calls to the source API return more (or less). Then the API would be called more slowly (or faster) to compensate.

Also we don't need to limit the number of ticks with take(3) now that we have throttling in place.

You can test the bitcoin ticker on Scastie remember to edit it to display your favourite currency pair and your favourite rounding.

Imparare ad usare Array.map in JavaScript

Andrea — Sat, 09 Jan 2021 11:24:46 +0000

Imparare ad usare Array.map in JavaScript

Il metodo .map è disponibile in JavaScript per tutti i browser e Node.js da ormai un bel pezzo (ECMAScript 2015). Se l’hai visto in giro e non lo stai ancora usando, dopo aver letto questo articolo nessuno ne saprà più di te sull’argomento!

Ma perché? Serve veramente?

Per velocizzare l’apprendimento, niente è meglio di un esempio. Qualsiasi programmatore JavaScript avrà visto questo codice svariate volte:

const users = [
  { email: 'giovanni@gmail.com', display: 'giova38' },
  { email: 'giangiorgio@toptech.com', display: 'giangi23' }
]

let usersWithAdmin = []
for (let i = 0; i < users.length; ++i) {
  const domain = users[i].email.split('@') // vedi nota 1
  let admin = false
  if (domain === 'toptech.com') {
    admin = true
  }
  usersWithAdmin[i] = Object.assign(users[i], { admin })
}
console.log(usersWithAdmin)
// l’output sarà
// [ { email: 'giovanni@gmail.com', display: 'giova38', admin: false },
//   { email: 'giangiorgio@toptech.com', display: 'giangi23', admin: true } ]

In questo codice creiamo un array vuoto, iteriamo tutti gli elementi di users e per ognuno controlliamo se il dominio della mail è "toptech.com", in quel caso dichiariamo l’utente come admin (tralasciamo la sicurezza di tale codice 🤣).

Una bella pappardella, lo stesso identico risultato si può ottenere con 4 leggibilissime righe.

const isUserAdmin = user => user.email.split('@')[1] === 'toptech.com'

const usersWithAdmin = users.map(user => 
  Object.assign(user, { admin: isUserAdmin(user) }))

Nella prima riga dichiariamo, in maniera un po’ più succinta, cosa significa essere admin. Nella seconda prendiamo un bel respiro. Infine, nelle ultime due "mappiamo" ogni elemento dell’array ad una copia con il nuovo campo admin aggiunto.

(Semi)immutabilità
Prima di spendere due parole su .map si prega di notare che usersWithAdmin prima era stato dichiarato let, nel nuovo codice è const¹. Questo ha implicazioni importanti: il nostro array nasce già inizializzato, non possiamo mai referenziare, nemmeno facendolo apposta, il nostro array in fase di inizializzazione.

Come funziona map

In due parole .map è un metodo che permette di applicare la stessa funzione a tutti gli elementi dell’array.

const array = ['ciao', 'mondo']
const maiuscolizza = stringa => stringa.toUpperCase()
console.log(array.map(maiuscolizza))
// l’output sarà
// [ 'CIAO', 'MONDO' ]

Nel caso sopra abbiamo usato la funzione maiuscolizza che manda una stringa nella sua versione maiuscola.

Guardiamo la segnatura della funzione che possiamo passare:

function callback(cursore[, indice[, array]])

Il cursore che è l’unico parametro usato negli esempi è l’elemento corrente che vogliamo elaborare.
indice è l’attuale indice associato al cursore. Il primo elemento avrà indice 0.
array ci restituisce l’array su cui stiamo agendo. Utile nel caso in cui la nostra funzione deve combinare diversi elementi. Da notare che si tratta dell’array originale.

Estrarre proprietà di un oggetto con map

Supponiamo di avere oggetti con molte proprietà e di volere un nuovo array con solo una di queste, l’equivalente di una colonna. Potremmo fare così:

const esperimenti = [
  { nome: 'Large Hadron Collider', costo: 5000000000 },
  { nome: 'Vulcano di bicarbonato', costo: 3.50 },
  { nome: 'International Space Station', costo: 150000000000 }
]
const nomiEsperimenti = esperimenti.map(esperimento => esperimento.nome)
console.log(nomiEsperimenti)
// l’output sarà
// [ 'Large Hadron Collider', 
//   'Vulcano di bicarbonato',
//   'International Space Station' ]

Reduce
Di solito si estrae una colonna per effettuare dei calcoli, ad esempio se vogliamo calcolare il costo totale degli esperimenti potremmo scrivere

esperimenti.map(e => e.costo).reduce((prec, succ) => prec + succ)
// output: 155000000003.5

Abbiamo usato reduce che agisce come accumulatore di valori ed effettua la somma. Piu informazioni sul sito di MDN.

Andrea Passaglia

Ad essere perfettamente onesti, const viene qui usato solo in maniera convenzionale a dire "ehy questo array non dovrebbe più essere mutato". La verità tuttavia è che, anche se non possiamo più riassegnare la variabile, l’array può benissimo essere mutato tramite assegnamento dei suoi elementi. Nei prossimi articoli risolveremo anche questo problema. ↩

10 days with the ZIO 2/10

Andrea — Mon, 27 Jan 2020 14:29:23 +0000

This miniseries aims at teaching how to use the ZIO at a very relaxed pace. Today we are going to build a word count app.

This means that given a text we want to know how many times each word appears.

Sneak Peek

Suppose you have the text

How much wood would a woodpecker peck if a woodpecker would peck wood?

The program should output

(a,2), (peck,2), (woodpecker,2), (would,2), (if,1), (wood,1), (much,1), (How,1), (wood?,1)

Setup

Like last time we are going to use Mill instead of SBT. Why? because it’s easier and takes 5 seconds.

Create a folder zio-word-counter
Inside it create a build.sc file with the following content

import mill._, scalalib._

object WordCounter extends ScalaModule {
  def scalaVersion = "2.13.0"
  val zioVersion = "1.0.0-RC17"
  def ivyDeps = Agg(
    ivy"dev.zio::zio:$zioVersion",
    ivy"dev.zio::zio-streams:$zioVersion"
  )
}

And we are done. We will need zio-streams as well since we are going to treat our text like a stream of words.

A simple stream of words

Create a folder named WordCounter (it has to match the object name we defined earlier) and a src folder nested inside. From there we can create a file named Main.scala with the following empty canvas.

// WordCounter/src/Main.scala
import zio._
import zio.console._
import zio.stream._

object Main extends App {
  def run(args: List[String]): ZIO[ZEnv, Nothing, Int] =
    putStrLn("hello world").map(_ => 0)
}

The first thing that we want is to have a decent text source. A string variable will do.

val someText = "How much wood would a woodpecker peck if a woodpecker would peck wood?"

And the run function can become like this

def run(args: List[String]): ZIO[ZEnv, Nothing, Int] =
  (for {
    chunk <- Stream(someText).run(ZSink.splitOn(" "))
    _ <- putStrLn(chunk.mkString(", "))
  } yield ()).map(_ => 0)

You can run this with the command mill --watch WordCounter (the watch will recompile and run when you save the file).

You should see the output

How, much, wood, would, a, woodpecker, peck, if, a, woodpecker, would, peck

The last word wood? is not printed because it doesn’t end with a space. You can add a space at the end to include it if you want; I will add one before the question mark.

Counting the words

We have our stream of words we need now to do two simple operations. The first operation is to pair words that are the same, the second is to count them.

The plan is to create one stream per word, since new streams are created and tracked by their key, setting their key to the word itself will work.

Counting the words will be done by collecting all the words in a list per stream. At the end we can just count the length of the list that will tell the number of words for that particular word.

streamOfWords.groupByKey(identity) { case (word, stream) =>
  stream.aggregate(ZSink.collectAll[String]).map(list => (list.size, word))
}

The identity function tells the groupByKey to use the word itself as the key. Then, for every stream we create a tuple of type(Int, String) that contains the count of the word and the word itself.

The complete run function looks like this.

def run(args: List[String]): ZIO[ZEnv, Nothing, Int] =
  (for {
    chunk <- Stream(someText).run(ZSink.splitOn(" "))
    counts <- Stream.fromChunk(chunk).groupByKey(identity) { case (word, stream) =>
      stream.aggregate(ZSink.collectAll[String]).map(list => (list.size, word))
    }.runCollect
    _ <- putStrLn(counts.mkString(", "))
  } yield ()).map(_ => 0)

The only differnte being that we created a stream from the chunk of words and printed the count instead.
Running this will print

(2,would), (2,woodpecker), (2,peck), (1,How), (2,a), (2,wood), (1,much), (1,if)

Simple to Complex

Admittedly we didn’t accomplish a lot. Just counting a few words is not really impressive. How about a word count of a sizeable file like the Divina Commedia? Let’s stash it in our tmp folder for later.

wget http://www.gutenberg.org/files/1012/1012-0.txt --output-document /tmp/divina.txt

First we need to read the file. To do it correctly we will take the name of the file from the command line like so

args.headOption
  .map(filePath => IO(Paths.get(filePath)))
  .getOrElse(IO.fail("No file specified"))

This means that if the argument is not passed the entire program will fail with an error that is represented by the string No file specified. We are not printing the error anywhere now. To do it we need to adapt the code after the yield.

def run(args: List[String]): ZIO[ZEnv, Nothing, Int] =
  (for {
    file <- args.headOption
      .map(filePath => IO(Paths.get(filePath)))
      .getOrElse(IO.fail("No file specified"))
    // ... other code ...
  } yield ())
    .onError(failure => putStrLn(failure.toString))
    .fold(_ => 1, _ => 0)

Note that we turned the map into a fold that will return 1 to the operating system in case of failure.

After getting the file we create a new InputStream which is the preferred way to read binary files.

inputStream <- IO(Files.newInputStream(file))

But this is a text file! 😱

We will have to turn the pesky bytes into utf8 character.

counts <- ZStream
  .fromInputStream(inputStream)
  .chunks
  .aggregate(ZSink.utf8DecodeChunk)
  // ...

Notes on the notes on the code

Warning From version 1.0.0-RC19 on all streams work by chunk so the above code and below notes are outdated. Just delete chunks and use a ZTransducer insted of a ZSink and it should just work.

Notes on the code

Now I need your attention for a second as it’s not completely clear what is happening in the above code. The ZStream.fromInputStream returns a StreamEffectChunk type. This means that it’s a stream (the Stream part, duh! 🤪) that it emits thunks (the Effect part) and works with chunks of bytes instead of single bytes. By default the size of the chunks is 4096 but it can be set manually.

The reason it emit thunks (lazily evaluated functions) is because it will read the next chunk of bytes only when the downstream is requesting them. This allows to read files that are bigger than the computer memory by doing it chunk by chunk.

Finally, the reason it uses chunks instead of single bytes is because of speed. Reading 4096 bytes all at once is order of magnitudes faster than reading them one by one. (Why 4096? Usually the file systems can’t access less than that, so if you set it to 1 for example, you are really accessing 4096 bytes and throwing away 4095 of them).

Next we have a choice to make, we can either work with the chunks or with the single bytes. By writing .chunks we cast to StreamEffect with Chunk[Byte] as emitted type. The alternative is .flattenChunks which casts as well to StreamEffect but now the emitted type is Byte.

We go with .chunks both for performance reasons and because there are some built-in functions that only work on chunks. The first one is aggregation with ZSink.utf8DecodeChunk.

Counting Words for serial now!

So we now are at a point in which we can generate Strings from the chunks of bytes. The next step is to separate the words. To do it we write

counts <- ZStream
  .fromInputStream(inputStream)
  .chunks
  .aggregate(ZSink.utf8DecodeChunk)
  .aggregate(ZSink.splitOn(" "))
  .flatMap(ZStream.fromChunk)
  .filter(_.matches("[\\w]+"))
  // ...

The splitOn unfortunately returns chunks so we “unchunk” them with .flatMap(ZStream.fromChunk). What this does is concatenating all the chunks in strict order and emitting the strings inside them one by one.

Next we filter the strings to make sure they are actual words. The [\w]+ regex basically means “letters of the alphabet and nothing else”.

The last thing we need to do is count the words! By recycling the previous code we get the following program

def run(args: List[String]): ZIO[ZEnv, Nothing, Int] =
  (for {
    file <- args.headOption
      .map(filePath => IO(Paths.get(filePath)))
      .getOrElse(IO.fail("No file specified"))
    inputStream <- IO(Files.newInputStream(file))
    counts <- ZStream
      .fromInputStream(inputStream)
      .chunks
      .aggregate(ZSink.utf8DecodeChunk)
      .aggregate(ZSink.splitOn(" "))
      .flatMap(ZStream.fromChunk)
      .filter(_.matches("[\\w]+"))
      .groupByKey(identity) {
        case (word, stream) =>
          stream
            .aggregate(ZSink.collectAll[String])
            .map(list => (list.size, word))
      }
      .runCollect
    _ <- putStrLn(counts.sortBy(_._1).reverse.mkString(", "))
  } yield ())
    .onError(failure => putStrLn(failure.toString))
    .fold(_ => 1, _ => 0)

The only thing I did is sorting the results by occurrence.
Launching this (mill --watch WordCounter /tmp/divina.txt) will take about a minute ☕️ and will print something like:

(3616,e), (3535,che), (2260,la), (1859,a), (1826,di), (1339,non), (1319,per), (1123,in), (1042,si), (777,le), (752,li), (736,mi), (652,il), (623,con), ...

Yes it’s a lot of work but a minute seems really long. I couldn’t figure out why it was so slow but I suspect the groupByKey is optimized to work with a few of streams, not thousands.

A faster alternative

A small modification will yield a much faster result (⚡️about 50x faster⚡️)

def run(args: List[String]): ZIO[ZEnv, Nothing, Int] =
  (for {
    file <- args.headOption
      .map(filePath => IO(Paths.get(filePath)))
      .getOrElse(IO.fail("No file specified"))
    inputStream <- IO(Files.newInputStream(file))
    counts <- ZStream
      .fromInputStream(inputStream)
      .chunks
      .aggregate(ZSink.utf8DecodeChunk)
      .aggregate(ZSink.splitOn(" "))
      .flatMap(ZStream.fromChunk)
      .filter(_.matches("[\\w]+"))
      .fold(HashMap.empty[String, Int])(
        (map, word) =>
          map.updatedWith(word)(
            maybeCounter => maybeCounter.map(_ + 1).orElse(Some(1))
          )
      )
      .map(_.toList)
    _ <- putStrLn(counts.sortBy(_._2).reverse.mkString(", "))
  } yield ())
    .onError(failure => putStrLn(failure.toString))
    .fold(_ => 1, _ => 0)

I swapped the groupByKey with a fold and sorted by occurrence (which is now _2).
Maybe with older versions of scala it would be even faster with mutable hash maps, but that would not be functional and thus not cool nor hip. I’m not even gonna try.

Conclusion

This was a short journey in very simple streaming applications. The zio-stream code is still a bit clunky in my opinion but it clearly shows a lot of potential and will surely be more powerful than both fs2 and Akka Streams. In the next tutorial we are going to do some concurrent functional programming.

10 days with the ZIO 1/10

Andrea — Fri, 13 Dec 2019 13:09:59 +0000

I’m always in pursue in ways to scale down software development. Why? Because to scale up you need to scale down!

When the size of the software system you are building goes up it becomes critical to have fast feedback loop, to be able to prototype new features and to debug quickly. You therefore need scalable tools in both ways.

Say Hello World to the ZIO

The ZIO is a functional library to program in Scala that scales very well in both directions. In this mini series we are going to spend some time with the ZIO building 10 small projects. We will start with a Hello World program, but first a digression on Mill.

Mill vs SBT

I like SBT, I think it’s much better than Maven and Gradle. But I think it’s common opinion that it’s a fairly complex machine. It’s the poster child of an era in which everyone was coming up with his own DSL and exploiting all the most powerful scala features. Since you have to basically learn another language to build a scala project, it means you have to spend a lot of time coming up with the right solution for your particular problem with specialized SBT knowledge. Lastly it’s quite slow.
Mill it’s an underrated build tool which resembles regular scala programming. It’s fairly easy to write a build.sc (equivalent to build.sbt) off the top of your head.
To install it just brew install mill or the equivalent for your operating system.

A blank slate

Let’s create a new folder hello-world-zio and inside it create a build.sc file with the following content.

// hello-world-zio/build.sc
import mill._, scalalib._

object zioHelloWorld extends ScalaModule {
  def scalaVersion = "2.13.0"
  val zioVersion = "1.0.3"
  def ivyDeps = Agg(
    ivy"dev.zio::zio:${zioVersion}"
  )
}

This means that we are creating a scala module called zioHelloWorld with the ZIO as dependency. The expected folder structure is very easy to remember

hello-world-zio/zioHelloWorld/src

Hello World from the ZIO

We can create the expected file at the above folder Hello.scala. We can put inside the simplest ZIO app I can think of:

import zio._, zio.console._

object Hello extends App {
  def run(args: List[String]) = putStrLn("Hello World").exitCode
}

Note that the App is really zio.App not the App provided with scala. The run has a slightly different signature. putStrLn comes from the Console module and by default it just prints to standard output. The final exitCode is a convenience method that maps a success of the program to a 0 to the operating system or a 1 in case of an error. Many number correspond to a different error, here is a link to a comprehensive list of errors)

Run, run. Click on run!

To run our program you can use the command

mill zioHelloWorld

You should see Hello World printed at the end of what seems to be some logging by mill itself.

Here is a working version on Scastie if you didn't follow along.

Conclusions

We have demonstrated how to the ZIO (with Mill) can easily scale down to a Hello World. We are going to use this minimalist template to build on top of it. Next up will be a word counter application.

Day 2

Go to day 2

DEV Community: Andrea

RTO, Return To On-premise

How we got there

Agency & Fate Sharing

What Changed?

Software Engineering for Xenophorms

Tilt: Improving Local Development with Production-Like Environments

The Problem Tilt Solves

Implementation Example

The Key Advantage: Instant Updates

Conclusion

Webhooks: A New Approach to an Old Problem

I Accidentally The Whole Cloud... And It Didn't Matter

The machines

The machines, as code

Cloud-init

Doing the stuff

Spinning up the wordpress

Bidirectional topic mirroring

Setting Up Bidirectional Mirroring with Pulsar

Start the Pulsar Clusters

Connect the Clusters

Create Tenants and Namespaces

Topic Creation

Testing the Setup

Extra

A closer look at the lifecycle of a send block in the Nano network

How does a send block get memorized in the Nano network

Conception

Another node receive the message

Processing message entries

Block processing

Conclusion

Composing complex programs with ZIO

The art of type chasing

Cat and mouse chasing

Creating a variable

Digital mouse that runs away

Digital cat that chases

Putting it all together

Environment channel

Failure channel

Success channel

Explanation of compositeProgram

Bonus section

Building a Bitcoin price ticker

Streams

Building a Bitcoin price ticker

Design

Programming

Imparare ad usare Array.map in JavaScript

Imparare ad usare Array.map in JavaScript

Ma perché? Serve veramente?

Come funziona map

Estrarre proprietà di un oggetto con map

10 days with the ZIO 2/10

Sneak Peek

Setup

A simple stream of words

Counting the words

Simple to Complex

Notes on the notes on the code

Notes on the code

Counting Words for serial now!

A faster alternative

Conclusion

10 days with the ZIO 1/10

Say Hello World to the ZIO

Mill vs SBT

A blank slate

Hello World from the ZIO

Run, run. Click on run!

Conclusions

Day 2

Explanation of `compositeProgram`