DEV Community: Javier Garcia

WTF are streams?

Javier Garcia — Tue, 04 Jun 2019 16:57:36 +0000

Streams have always been a complicated topic, but with a little help from some examples, hopefully we can make it right today.

Streams are, in essence, a sequence of data elements which are available over time. Just like an actual stream of water, the data flows/becomes available, instead of having it all from the beginning. There are many benefits to this although the two most important are a huge enhancement to performance and the fact that data is not always available immediately.

Like just pointed, one of the main reasons to use streams is because sometimes, data is not immediately available. For example, if you're listening to a weather streaming API, the current temperature is calculated in the current moment, so (1) the data is infinite as long as you continue listening to the service, and (2) new data is available each minute, as it's produced. This can't be modelled with a finite amount of data, so that's when we decide to model it with an infinite stream of data.

Since the size of a stream is undefined, they are potentially unlimited, it's important to remember that we can't operate on them as a whole, like we're used to when working with lists or arrays. This way, functions which are applied to streams usually return another stream with the data modified. These are called filters, and when chained, they form pipelines.

Another benefit mentioned was performance. When processing very big amounts of data, this can usually end up in a very big memory hit to our computer. If you try to read a 20Gb file with data about cats, process it and then send it to a friend, that would mean that apart from whatever amount of memory our application uses, we're going to load an extra 20Gb to memory. Most laptops would die! But that doesn't mean it isn't doable.

If instead of reading the file as a whole we model it as a stream of data, we can read one line at a time, process those lines and send them, also as a stream. This would make our application just use an extra few bytes instead of the awful 20Gb. When thinking about streams, we always have to think about them as a nice sequence of manageable chunks of data our application can process – once it's finished with a chunk, it can read and process the next.

Simplified I/O over a String, in Elixir

Our first practical approach to streams is going to be the StringIO module in Elixir. A StringIO is not really an actual stream, but just a wrapper around a string that allows us to apply some of the standard I/O operations over streams to the string. For us it's going to be perfect because we can use it to familiarize with the operations:

Open: obtains exclusive access over the resource.
Read/Write: basically reads or writes chunks of date from/to the stream.
Close: returns the resources to the OS (operative system).

First, we open a string IO. From the response of the open/1 function, we can see that it gives us a reference to the stream.

iex>  {:ok, pid} = StringIO.open("content")
{:ok, #PID<0.470.0>}

If we want to inspect the contents of the stream, we can do so via the contents/1 function, by passing it the previously obtained reference. In elixir the contents/1 function will always return a tuple with the input buffer and the output buffer, such as {"in", "out"}. In this case, the output buffer will be empty since we haven written anything to it.

iex(1)> StringIO.contents(pid)
{"content", ""}

Since StringIO is a a wrapper which models the string as a stream, we can use the standard functions to read and write to streams from the IO module. In this case, to write some content, we can use the write/2 function. Notice how we now have data both in the input buffer and the output buffer.

iex(2)> IO.write(pid, "written")
:ok

iex(3)> StringIO.contents(pid)
{"content", "written"}

Most stream modules in most languages also give us a way to flush the content, which means it forces any bytes in the stream to be written out. This applies to the output buffer.

iex(4)> StringIO.flush(pid)
"written"

iex(5)> StringIO.contents(pid)
{"content", ""}

Lastly, if we want to read from the input buffer, we can use the read/2 function, thus emptying the stream of data:

iex(6)> IO.read(pid, :all)
"content"

iex(7)> StringIO.contents(pid)
{"", ""}

Notice how in this specific case Elixir models the StringIO as a tuple with both an input buffer and and output buffer, a buffer to which we can write and one from where we can read from.

I/O on a File, in C#/.NET

Moving to a more practical example, we're going to check out how to work with files in streams. When working with files, opening and closing the streams starts to take much more importance, but we'll discuss that later.

In .NET land, in order to create a file, we can make use of the File.Create function. This will provide us with a FileStream which models our file, so we can write to it. Once we open the stream and write to it, we will have to close it in order to persist those changes and free the resources the OS has given us. Furthermore, to read the content again, we will reopen another stream with File.OpenRead and read byte by byte. The snippet looks as follows, credits to the MSDN:

using System;
using System.IO;
using System.Text;

namespace StreamTime
{
    public class FileTheStream
    {

        public static void Main()
        {
            const string path = @"/Users/jgarcia/Desktop/example.txt";

            //Create the file.
            using (FileStream fs = File.Create(path))
            {
                AddText(fs, "This is some text");
                AddText(fs, "This is some more text,");
                AddText(fs, "\r\nand this is on a new line");
                AddText(fs, "\r\n\r\nThe following is a subset of characters:\r\n");

                for (var i = 1; i < 120; i++)
                {
                    AddText(fs, Convert.ToChar(i).ToString());
                }
            }

            //Open the stream and read it back.
            using (FileStream fs = File.OpenRead(path))
            {
                var b = new byte[1024];
                var temp = new UnicodeEncoding();
                while (fs.Read(b,0,b.Length) > 0)
                {
                    Console.WriteLine(temp.GetString(b));
                }
            }
        }

        private static void AddText(Stream fs, string value)
        {
            var info = new UnicodeEncoding().GetBytes(value);
            fs.Write(info, 0, info.Length);
        }
    }
}

As you can see, code-wise, we're just opening a stream, reading, writing, and closing it. Well, you might be wondering where the closing is happening. Take into account that in C#, whenever we use the using keyword with a resource, once it's finished using it, it closes the resource – Just like Java's try with resources. But let's talk further about the opening and closing of streams.

Why do streams have to be opened or closed?

We mentioned that every time we open a stream the OS needs to dedicate resources, but we never talked about which, or how. Most of the time, it depends on the nature of the stream which we have opened – it's not the same to open a socket than to open a file, etc. For the time being, we can concentrate on the files.

Whenever a new file stream is open, the OS dedicates a file descriptor, commonly known as a file handle, to the application. A file descriptor basically is a number that uniquely identifies an open file in a computer's OS – it describes a data source, and how it can be accessed. File descriptors point towards the kernel's global file table, which contains information such as the offset and the access restrictions of the stream.

As you can imagine, file descriptors isn't like memory – if you don't return them to the OS the situation can grow ugly fast. It's just a matter of time until the application crashes in a long running application. This is usually called a file-handle leak. In Windows computers, when you try to delete a file, usually one of the consequences is that it says it's being used by another program. The resource hasn't been freed properly.

Luckily, most languages nowadays provide us with constructs which allow us to free those resources appropiately. Like mentioned, we have statements like using in C# or try with resources in Java in our toolbox. In Elixir though, you have to close it with the IO.close/1 function.

I/O over a socket, in Java

Another scenario where streams are used is when an application opens a socket. In order for two processes to communicate, they each need to open a socket, and then send their messages through there – sockets send and receive messages through streams. Showing up next is a snippet which contains a very simple echo server developed in Java. Notice how the socket contains both an input stream and an output stream, and how we read data the data sent through the input stream, to then write it to the output stream.

package com.manzanit0;

import java.io.*;
import java.net.ServerSocket;

public class EchoServer {
    public static void main(String[] args) {
        int portNumber = 4098;

        try (
                var serverSocket = new ServerSocket(portNumber);
                var clientSocket = serverSocket.accept();
                var outputStream = clientSocket.getOutputStream();
                var inputStream = clientSocket.getInputStream()
        ) {
            while(true) {
                var bytes = readAllBytes(inputStream);
                outputStream.write(bytes);
            }

        } catch (IOException e) {
            System.out.println(e.getMessage());
        }
    }

    private static byte[] readAllBytes(InputStream stream) throws IOException {
        StringBuilder data = new StringBuilder();

        // available only returns a value after reading at least 1 character -> do/while.
        do {
            data.append((char) stream.read());
        } while (stream.available() > 0);

        return data.toString().getBytes();
    }
}

The other thing I want to mention is that since streams work with undefined amounts of data, potentially infinite, if you tried executing the program, you will have noticed that it doesn't stop running until you feed it an abort command (Ctrl/Cmd + C). We can continue inputing data for as long as we wont and the stream will continue feeding it. In this particular example I have elaborated a specific bytes[] readAllBytes(InputStream stream) function which reads only the available bytes and returns them, but the InputStream class provides us with a readAllBytes() method which blocks until the stream is closed, then returning all the bytes received.

You might be wondering though, what if I want to read the data of my stream a second time? Is it possible?. Indeed it's possible, but for understanding how, we must introduce one last concept: seeking.

Seeking a stream – understanding the side-effects of reading

If you've tried reading a stream a second time, you might have found yourself not being able to read previous data, but just reading new data. Some streams don't support seeking, but assuming they do, the reason behind this is that streams have a cursor which points to the last byte read. Every time we read a new byte, that cursor is advanced to the new position. In order to read already processed bytes we would need to rewind that cursor all the way to the beginning. This is called seeking.

In Java, the way to do this is using the mark(int) and reset() method of the InputStream class, in the C# example, we would simply set file.InputStream.Position = 0. These are the side effects of reading a stream. If a stream doesn't support seeking, another solution would be copying our read bytes to another array and maintain a copy. Nonetheless, take into account that sometimes one of the purposes of using streams is to go easy on memory consumption, and we're copying all the read data in a memory array, then we're annulling this completely.

Wrapping up

We've covered a lot of things, but if we were to do a really quick TLDR and summarize some of the key points, I would just go with:

Streams manage undefined amounts of data – sometimes infinite.
Streams enable a huge performance boost for applications since they don't have to load all the data in memory before processing it.
Just as streams have to be opened, they must be closed, otherwise the OS' resources aren't freed and that can potentially be a very expensive cost, be it sockets or file handles.
Some languages provide constructs to handle the disposal of resources, like using in C# or try in Java.
Streams have a cursor which points to the last read byte. In some cases, we can rewind that cursor in order to read the same data a second time. This is called seeking.

Following up, in case you want to delve a little more into sockets, as a concept, feel free to check out this other post I have about them: URL.

Originally posted at: https://manzanit0.github.io/

Understanding Genservers

Javier Garcia — Tue, 21 May 2019 13:04:03 +0000

If you're reading this, it probably is that you've had to use Elixir's Genserver behaviour already but you're wondering how it works. For me, the first time I tried using it I got it completely wrong – Instead of using it as a store of data I ended up spawning up a brand new Genserver for each new record I wanted to save. When I found out, I realised it was simply because I wasn't really understanding what was going on under the hood.

What is a Genserver?

A Genserver, according to the hexdocs is:

A behaviour module for implementing the server of a client-server relation.

A GenServer is a process like any other Elixir process and it can be used to keep state, execute code asynchronously and so on. The advantage of using a generic server process (GenServer) implemented using this module is that it will have a standard set of interface functions and include functionality for tracing and error reporting. It will also fit into a supervision tree.

At the end of the day it's simply an abstraction for client-server behaviour: it allows us to define a module of our own with a set of callbacks and we can call it from anywhere in our code base. It basically serves two different purposes: first it allows us to execute pieces of code asynchronously, so we don't necessarily block our main thread, and secondly it helps us save state.

In order to help us wrap our heads around this concept, we're going to build our own implementation of a Genserver in Elixir. A really simple one!

Processes 101

Before getting started, it's important to review how processes work and communicate in Elixir. In Elixir, all code runs inside isolated processes. The basic mechanism for spawning those process is the spawn/1 function.

iex> spawn fn -> "Hello world!" end
#PID<0.117.0>

As you can see, spawn/1 returns a process identifier (PID). Spawned processes usually execute the function we have provided them with and then exit, so it's most likely dead. Take into account that wee can also spawn processes with functions from already existing modules with spawn/4, but we'll check it out later in the examples.

Furthermore, in order for processes to communicate Elixir provides us with the send/3 and receive/1. send/3 allows us to send a message to any process, given that we know it's PID, and receive intercepts all the messages in the current process. Check out this example:

iex> send self(), {:msg, "Hey!"}
{:hello, "world"}
iex> receive do
...>   {:msg, msg} -> msg
...>   _ -> "not a message!"
...> end
"Hey!"

Notice that we are sending a tuple to self(), which is a reference to the current process and once that is sent, we intercept it by running a receive/1 clause.

Now, taking all this into account we could probably write two simple modules which exchange messages in the following way:

defmodule Person do
    def create, do: spawn(Person, :listen, [])

    def say(to, message), do: send(to, {:msg, message})

    def listen do
        receive do
            {:msg, msg} -> IO.puts("Said '#{msg}' to #{inspect self()}")
        end
        listen()
    end
end

And on iex>:

iex> marta = Person.create()
#PID<0.134.0>
iex> Person.say(marta, "Hey peep!")
Said 'Hey peep!' to #PID<0.134.0>
{:msg, "Hey peep!"}"

Storing state in a stateless world

So far, in our journey to understand Genservers, we have learned about spawn/1, send/3 and receive/1, but, how does that take us any closer to understanding how Genservers work?. Before jumping to the Genserver behaviour we have still one more piece of the puzzle to unveil: state.

As you know, Elixir is a functional language which has now knowledge of state as is – we create modules, which have functions, and we give them data, which they spit out processed. But we don't have instances as we would in C# or Java, instances that store state for us. Furthermore, when studying processes, as we spawn them, they die. So how can we store state? The answer is recursion.

In order to be able to maintain some state in Elixir, the common pattern is for a process to recursively call itself with the state it has. Take a look at the following example:

defmodule Counter do
    def init, do: loop(0)

    def loop(counter) do
        IO.puts counter
        loop(counter + 1)
    end
end

If you try to execute it...

iex> Console.init()
... (many numbers which I will not paste)
199943
199944
199945
199946
199947
199948
199949

BREAK: (a)bort (c)ontinue (p)roc info (i)nfo (l)oaded
       (v)ersion (k)ill (D)b-tables (d)istribution
^C

Yes, it starts printing all the numbers and won't stop until you stop the process, which you can do by pressing twice Ctrl + C. Anyways, as you've seen, we have been able to store and change state with a recursive loop. That is the key to how Genservers will hold state. Now, let us move forward to the real deal!

A homemade Genserver

With all the tools we have gathered, we can now commence. Since I'm all about TDD, let's start with a test. Sticking to the type of APIs Elixir provides us, I will want an init/0 function which will spawn our server for us, and a save/2 function for saving the message to the server. This is my test:

defmodule StoreTest do
  use ExUnit.Case

  test "saves message" do
    pid = Store.init()
    response = Store.save(pid, "Hey!")

    assert {:ok, "message received!"} == response
  end
end

In order to make that test pass, we kind of do have to write a little bit of code. First we need the init/0 function, but we also need the loop we talked about which will be storing the state for us and a way to send back the response. If you've tinkered around with Genservers a little, you will know that they allow the consumers to send both synchronous and asynchronous messages via call/2 and cast/2. In this case, we're trying to develop a function similar to call/2 – a function which waits for the server to create the response and return it.

First, the init/0 function will use spawn/4 which we mentioned at the beginning of the post, it will allow us to pass it a function defined in one of our modules:

  def init() do
    spawn(__MODULE__, :loop, [[]])
  end

The third parameter are the args for the function, in this case we want it to be an empty list, so that explains the [[]].

Continuing forward, we want our save/2 function to send a message to the server and await the response, so that should be fairly straightforward with send/3 and receive/1:

def save(pid, message) do
    send(pid, {:save, self(), message})

    receive do
        {:ok, response} -> {:ok, response}
    end
end

Lastly, the loop. When we spawn the process, we're invoking a loop/1 function which receives the state and is supposed to keep the wheel going. Since our save/2 function sends a message to the server and awaits the response, we can kind of assume that the loop will be waiting for a message and then sending a response back.

def loop(state) do
    state =
      receive do
        {:save, from, msg} ->
          send(from, {:ok, "message received!"})
          [msg | state]
      end

    loop(state)
end

And if we put this all together...

defmodule Store do
  def init() do
    spawn(__MODULE__, :loop, [[]])
  end

  def save(pid, message) do
    send(pid, {:save, self(), message})

    receive do
      {:ok, response} -> {:ok, response}
    end
  end

  def loop(state) do
    state =
      receive do
        {:save, from, msg} ->
          send(from, {:ok, "message received!"})
          [msg | state]
      end

    loop(state)
  end
end

If we now run mix test we should have a nice green output. Next, we want a function to retrieve all the messages we've stored. I start, as always, with the test:

test "retrieves all message" do
    pid = Store.init()
    Store.save(pid, "...world!")
    Store.save(pid, "Hello")

    response = Store.fetch(pid)

    assert {:ok, ["Hello", "...world!"]} == response
end

The good thing is, in this case, we just need to develop the fetch/1 function. In this case, fetch/1 will look very similar to save/2, in the sense that it will send a message and expect a response. The core of the logic will be coded in the loop – we need to make the server respond with all the data. Once we finish, our code looks like this:

defmodule Store do
  def init() do
    spawn(__MODULE__, :loop, [[]])
  end

  def save(pid, message) do
    send(pid, {:save, self(), message})

    receive do
      {:ok, response} -> {:ok, response}
    end
  end

  def fetch(pid) do
    send(pid, {:fetch, self()})

    receive do
      {:ok, response} -> {:ok, response}
    end
  end

  def loop(state) do
    state =
      receive do
        {:save, from, msg} ->
          send(from, {:ok, "message received!"})
          [msg | state]
        {:fetch, from} ->
          send(from, {:ok, state})
          state
      end

    loop(state)
  end
end

What about the async functions?

As I commented before, Genservers also have asynchronous handlers: cast/2. The reason why I decided to only implement the synchronous handlers is because the async ones are simpler. Our synchronous function looks like this:

def fetch(pid) do
    send(pid, {:fetch, self()})

    receive do
      {:ok, response} -> {:ok, response}
    end
end

Yet, if we want to make it async, we just have to delete the receive/1 clause.

def fetch(pid) do
    send(pid, {:fetch, self()})
end

Conclusions

After having read through the whole thing, Genservers don't look so dangerous anymore, do they? At the end of the day they are, like our Store module, a simple wrapper around processes which communicate between each other and provide us with a client-server architecture. Before finishing though, I will mention Agents.

Agents are yet another abstraction provided to us by the Elixir core team to make working with state easier. At the end of the day, they are simply a wrapper around Genservers themselves, but they provide us with a much cleaner and easier API to use – we don't have to worry about implementing the cast/call callbacks boilerplate. Next time you need to store state in your application, give it a thought – Do you need a Genserver? Can you do it with a Task or an Agent instead? It's always about simple code!

Originally posted at https://manzanit0.github.io

Automating Salesforce deployments with PowerShell

Javier Garcia — Wed, 15 Aug 2018 22:45:22 +0000

Recently I've been learning a little bit of PowerShell, initially for the sake of automating some tedious tasks I had to do at work like calling a REST endpoint 10-20 times to populate some DB for testing or simply pipelining some CLI commands.

The thing is, as I got more comfortable with PS, I started to realize the huge amount of things that can be done with scripts which make our day to day easier. The biggest one that came to my mind was Salesforce deployments.

Usually, once we've finished developing a certain feature in Salesforce, we run all tests, pull all our changes from the environment where we were developing (In Salesforce, you develop remotely, not locally), deploy to a staging environment and run again the tests there.

The issue is that that process involves a lot of manual steps. So I decided to try and automate it. For it, I looked into the SFDX CLI and tried to reuse most of its features: an authentication command, the actual command which calls the deployment endpoint, etc.

Eventually, after a few hours, the result was this:

It's a little nifty script which prompts a tab in your default browser so you log into the destination org and pipelines the deployment and the running of tests with real-time status in the console. That I did simply by polling the API.

There are, of course, many things to be improved in the script but, as a starter, it's already saving me a lot of time! :)

I'm a Salesforce Technical Lead, Ask Me Anything!

Javier Garcia — Mon, 13 Aug 2018 11:15:05 +0000

I'm currently working for a Swiss company leading the development of their in-house Salesforce platform.

Since Salesforce is a stranger for most developers, I thought it could be interesting to do an AMA about it. Feel free to ask anything :)

Aquamon: Interacting with Salesforce Tooling API

Javier Garcia — Wed, 27 Jun 2018 11:40:03 +0000

Since I started working at Frontiers, we've had many technical tasks that were monotonous and repetitive, typical in most technical jobs. One of them was refreshing our Salesforce Sandboxes every time we wanted to start working on a new feature.

Our modus operandi usually is to first refresh a sandbox, copy from Production, then branch our repository's Test branch, push all the metadata to the sandbox, and the finally start working on it until it's finished and a pull request is made.

If you've worked sometime with Salesforce, you are probably aware of how tedious it is to refresh a sandbox from Production and, once it is finished, set all the environment variables in Custom Settings or Custom Metadata. Here is where Aquamon came into play.

After having to do this for a couple of months, I came to the conclusion that this had to be automated. For this, I made a small CLI application which simply connected to Salesforce via the Tooling API and sent a refresh process order with a PostSandboxCopy execution script.

The application is separated in 4 different modules: the first one and most simplistic is the post-copy script, developed in Apex, which lives in Salesforce. This is executed every time a Sandbox is refreshed and simply sets some default values to the custom settings, remote site settings, etc. It can also seed the DB to have some records to work with without having to manually create them.

The post-copy script can look something like this, from the docs:

The rest of the application, I developed in C# with .NET core. It has 3 modules: one for configuration, another one which manages all the connections/requests with the Salesforce Tooling API and a third for the CLI. Since I don't want to make this post extremely long, I'll dig deeper into the Tooling client and skip quickly over the others.

While the Tooling API is not specially complicated to use, there are some perks to it. It has a REST API and a SOAP API, but since we can do everything we need to with the REST API, I took that path.

Authenticating in Salesforce

Now, first is first, and to be able talk to the Salesforce APIs we first need to authenticate. In order to do this, I created a DTO which contains all the authentication data and a small client to do so.

Salesforce's APIs use OAuth, so we only need to POST to https://login.salesforce.com/services/oauth2/token and get the authentication token which we will later use in all future calls.

Connecting to the Tooling API

Once we're authenticated, in order to create, refresh and poll sandboxes, Salesforce exposes two different entities: the SandboxInfo and the SandboxProcess. To create or refresh sandboxes we would have to create a SandboxInfo record, to check the status, we would poll the SandboxProcess.

All our development is going to oscillate between these two entities: we want a new sandbox? Post a SandboxInfo. Do we want to refresh it? Patch it. Do we want to get the status? Then Get it.

To achieve this, I simply created a DTO for the SandboxInfo, based on the documentation and a SandboxManager class which would make the different calls to the different endpoints. If you notice, we inject the previous ForceClient which has already authenticated us.

Just as a heads up, I added a couple of extra methods to ForceClient.cs in order to encapsulate all the HttpClient instances.

What about the rest of the application?

Now, all the latter summarizes the interaction with Salesforce. To wire that up in a CLI application, I used the out-of-the-box Microsoft.Extensions.Configuration to read the configuration and bind it to different POCOs and Microsoft.Extensions.CommandLineUtils to set up the commands and argument parsing.

If you want to see the whole code, please do visit the github repository here.

On a side note, Microsoft.Extensions.Configuration is no longer being actively developed, so if I had to re-develop the application I would maybe go for a different option, but if you're interested in it, Nate McMaster has forked it here, which is nice.

TL;DR;

As you have seen, interacting with the Salesforce APIs is not very difficult, but it does require getting to know them. At the start it took me a little time to figure out how to query the system, and some other perks related to it like putting together the actual string. Nonetheless, once you get it rolling, it's very easy to keep pumping more stuff out.

Furthermore, I encourage you guys to download it, test it and feel free to give me any feedback! It's a tool I currently use on a daily basis and even though it is still not perfect, it does save me a bunch of time.