How do I manage my messaging flow?

Question

I have this Javascript code that represents my connection to the Websockets server, where it receives and sends messages to the server:

var connection = new WebSocket('ws://127.0.0.1:2333');

connection.onopen = function () { // abriu a conexão com o servidor 
};

connection.onmessage = function (e) { // recebe do servidor
};

function sendToServer(data) { // envia mensagens para o servidor
   connection.send(data)
}

How do I make sure that messages are not sent and received simultaneously, for example if the entrance gate onmessage is busy receiving messages from the exit gate sendToServer() will not send anything until all entries are completed. With these three criteria:

• No message should be lost
• Everything has to happen automatically
• The sequence must be respected

When I say that no message should be lost I say that if a server message comes while an output message is being worked on it will be "saved" and be "worked on" as soon as possible.

When I say that everything has to happen automatically I mean that nothing can stand still, if a message was saved to be executed as soon as possible, it has to be executed as soon as possible, it cannot be "forgotten"..

And finally the sequence, the sequence must be respected, but the sequence in which messages arrive/leave inside the script. Assuming a message A, coming from the server and then while the message A is "worked" comes another message from named server B and after B another from the called server arrives C and while all these, (A, B and C), wait in the queue to be "worked" the user orders that three other messages should be sent to the server, first D, afterward E and then F, in that order respectively.

Assuming all this, the final output/input or "job" sequence of this script should be:

A -> B -> C -> D -> E -> F

i think of it as an airport.. but it’s kind of complex for me. I tried to apply the semaphore but I did not succeed so I come to appeal..

"job" = send/receive to/from server

Answer 1

You need a queue: whenever the sendToServer is called, put the data in the queue instead of sending it straight. At the end, of the two:

• If the system is in a state that allows messages to be sent, send them at once (only one);
• Otherwise do not send.

When the system changes from "locked" to "available" status, send everything in the queue (so you don’t have to wait for the next call sendToServer.

var connection = new WebSocket('ws://127.0.0.1:2333');

    var fila = [];
    function esvaziaFila() {
        while ( fila.length > 0 ) {
            if ( !podeProcessar )
                return;
          
            var proximo = fila.shift();
            if ( proximo.envio )
                connection.send(proximo.dados);
            if ( proximo.recebimento )
                processar(proximo.evento);
        }
    }
    var podeProcessar = false;
    
    connection.onopen = function () { // abriu a conexão com o servidor 
        document.body.innerHTML += "<pre>Abriu a conexão</pre>";
        podeProcessar = true;
    };
    
    connection.onerror = function (error) { // ocorreu um erro na comunicação
    };
    
    connection.onmessage = function (e) { // recebe do servidor
        rec("cliente", e);
        fila.push({ recebimento:true, evento:e });
        esvaziaFila();
        
    };

    function processar(evento) { // recebe mensagens do servidor
      document.body.innerHTML += "<pre>Cliente vai processar " + JSON.stringify(evento) + "; não faz mais nada até estar pronto!";
      podeProcessar = false;
      
      tarefaDemorada(evento);
    }
    
    function sendToServer(data) { // envia mensagens para o servidor
       fila.push({envio:true, dados:data});
       esvaziaFila();
    }

/********** Mockups para testar ***************/
function rec(classe, dados) {
    document.body.innerHTML += "<pre class='" + classe + "'>" + classe + " recebeu: " + JSON.stringify(dados) + "</pre>";
}
function env(classe, dados) {
    document.body.innerHTML += "<pre class='" + classe + "'>" + classe + " enviou: " + JSON.stringify(dados) + "</pre>";
}

function WebSocket() {
  this.send = function(data) {
    rec("servidor", data);
  }
  // Manda algumas mensagens no futuro
  var self = this;
  setTimeout(function() { self.onopen() }, 500);
  setTimeout(function() { env("servidor","A"); self.onmessage("A") }, 2000);
  setTimeout(function() { env("servidor","B"); self.onmessage("B") }, 4000);
  setTimeout(function() { env("servidor","C"); self.onmessage("C") }, 6000);
}

setTimeout(function() { env("cliente","D"); sendToServer("D"); }, 8000);
setTimeout(function() { env("cliente","E"); sendToServer("E"); }, 10000);

function tarefaDemorada(evento) {
    setTimeout(function() {
        document.body.innerHTML += "<pre>Terminou de processar " + JSON.stringify(evento) + "; pode processar o resto.";
        podeProcessar = true;
        esvaziaFila();
    }, evento == "B" ? 5500 : 1250);
}

.servidor {
  color: blue;
}

.cliente {
  color: red;
}

Alternative: graph of dependencies

If you have a complex sending/receiving flow, one way to manage this complexity is to create a dependency graph between your messages, both incoming and outgoing. I suggest this - not a total order - as it is complicated to say that "event A occurred before event B" when A originated on a machine (e.g.: client) and B in another (e.g.: the server)^[1].

An example combining the queue technique with a dependency graph would be the following:

var fila = [];
function processaFila() {
    while ( fila.length > 0 ) {
        var proximo = fila[fila.length-1];

        if ( !proximo.pronto ) // Se está aguardando outro processamento
            break;             // Não processa mais nada

        fila.pop();
        if ( proximo.envio ) // Se está pronta pra ser enviada
             connection.send(proximo.dados);
        if ( proximo.recebimento ) // Se foi recebida e tem de ser processada
            proximo.processar(proximo.evento);
    }
}

function incluiNaFila(job) {
    var i = fila.length;
    while ( i > 0 ) {
        // Se o próximo da fila precisa ser executado depois da tarefa atual
        if ( dependeDe(fila[i-1], job) )
            break; // Coloca a atual no começo da fila

        // Senão, coloca a atual depois do primeiro da fila
        fila[i] = fila[i-1];
        i--; // Repete a lógica pro segundo da fila, etc
    }
    fila[i] = job;
}

connection.onmessage = function (e) {
    incluiNaFila({
        recebimento:true, pronto:true,
        evento:e,
        processar:function(e) { ... }, // Função que processa a entrada
        ... // outros campos que determinam a dependência
    });
    processaFila();
}

function sendToServer(data) {
    incluiNaFila({
        envio: true, pronto:true,
        dados: data,
        ... // Outros campos que determinam a dependência
    });
    processarFila();
}

function dependeDe(tarefaA, tarefaB) {
    /* Aqui entra sua lógica específica.
       Ela deve retornar false se a tarefa B pode ser executada depois
       de A (B é uma tarefa mais nova que A), ou true se A precisa
       esperar B estar pronta pra executar.
    */
}

After a quick read on your example in Pastebin (in the comments), I would suggest something like starting point:

var sequencial = 0; // Cada mensagem de iniciativa do servidor ou do cliente
                    // recebe um id sequencial; mensagens que são respostas
                    // a outra mensagem recebem o mesmo id da mensagem original.

connection.onmessage = function(e) {
    var id = sequencial++;
    incluiNaFila({
        recebimento:true,
        evento:e,
        processar:function(e) {
            var aguradando = {
                recebimento:true, processar:function(){},
                pronto:false, id:id
            }

            incluiNaFila(aguardando);
            funcaoquetrataamensagem(.......).then(function() {
                aguardando.pronto = true;
                processaFila();
            });
        },
        id:id
    });
    processaFila();
}

// Envia uma mensagem, sem impor nenhuma ordem específica (iniciativa do cliente)
function sendToServer(data) {
    incluiNaFila({
        envio: true, pronto:true,
        dados: data,
        id:sequencial++
    });
    processarFila();        
}

// Envia uma mensagem como resposta a uma mensagem recebida
function respondToServer(idMensagemOriginal, data) {
    incluiNaFila({
        envio: true, pronto:true,
        dados: data,
        id:idMensagemOriginal+1 // Garante que só será enviada quando a
                                // mensagem original tiver sido processada
    });
    processarFila();        
}

function dependeDe(a, b) {
    return ( a.id > b.id );
}

In this example the messages received and the messages sent on the client’s initiative will be processed in the order (queue), while messages that are reply to another message will be processed only after the original message has been fully processed. Any message that needs to "do more things" before finishing its processing blocks the entire queue - by inserting a new task into it aguardando with the same id, and hence the same priority, of the original message.

---

[1]: There are several attempts to solve this problem, but none perfect. Choosing a side of communication to be authoritative can cause starvation on the other; The technique Operational Tranform is promisory, but does not apply to all cases; in some scenarios (e.g., the bitcoin network), the Proof of work can be used to reduce the problem, but does not eliminate it altogether. Etc.

Answer 2

I don’t know if the Websocket object has some kind of control, but you can control it manually

var connection = new WebSocket('ws://127.0.0.1:2333');
var available = true;

connection.onopen = function () { // abriu a conexão com o servidor 
    // mantém available = false
};

connection.onerror = function (error) { // ocorreu um erro na comunicação
    available = true;
};

connection.onmessage = function (e) { // recebe do servidor
    processa(e);
    available = true;
};

function sendToServer(data) { // envia mensagens para o servidor 
    var interval = setInterval(function () {
        if (available) {
            clearInterval(interval);
            available = false;
            connection.send(data);
        }
    }, 1000)
}