Erlang Port Test

main.cpp

#include <ei.h>

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <io.h>

#include "marsharling.h"

#define BUF_SIZE 128 

/*-----------------------------------------------------------------
 * API functions
 *----------------------------------------------------------------*/
long add(long a, long b) {
  return a + b;
}

long multiply(long a, long b) {
  return a * b;
}

double divide(double a, double b) {
  return a / b;
}

/*-----------------------------------------------------------------
 * MAIN
 *----------------------------------------------------------------*/
int main() {
  byte*     buf;
  int       size = BUF_SIZE;
  char      command[MAXATOMLEN];
  int       index, version, arity;
  long      a, b, c;
  double    x, y, z;
  ei_x_buff result;

  _setmode(_fileno(stdout), O_BINARY);
  _setmode(_fileno(stdin), O_BINARY);

  if ((buf = (byte *) malloc(size)) == NULL) 
    return -1;
    
  while (read_cmd(buf) > 0) {
    /* Reset the index, so that ei functions can decode terms from the 
     * beginning of the buffer */
    index = 0;

    /* Ensure that we are receiving the binary term by reading and 
     * stripping the version byte */
    if (ei_decode_version(buf, &index, &version)) return 1;
    
    /* Our marshalling spec is that we are expecting a tuple {Command, Arg1, Arg2} */
    if (ei_decode_tuple_header(buf, &index, &arity)) return 2;
    
    if (arity != 3) return 3;
    
    if (ei_decode_atom(buf, &index, command)) return 4;
    
    /* Prepare the output buffer that will hold {ok, Result} or {error, Reason} */
    if (ei_x_new_with_version(&result) || ei_x_encode_tuple_header(&result, 2)) return 5;
    
    if (!strcmp("add", command) || !strcmp("multiply", command)) {
      if (ei_decode_long(buf, &index, &a)) return 6;
      if (ei_decode_long(buf, &index, &b)) return 7;

      if (!strcmp("add", command))
        c = add(a, b);
      else
        c = multiply(a, b);

      if (ei_x_encode_atom(&result, "ok") || ei_x_encode_long(&result, c)) return 8;
      
    } else if (!strcmp("divide", command)) {
      /* Allow incoming parameters to be integers or floats */
      if (!ei_decode_long(buf, &index, &a))
        x = a;
      else if (ei_decode_double(buf, &index, &x)) 
        return 6;
        
      if (!ei_decode_long(buf, &index, &b))
        y = b;
      else if (ei_decode_double(buf, &index, &y)) 
        return 7;
      
      if (y == 0.0) {
        if (ei_x_encode_atom(&result, "error") || ei_x_encode_atom(&result, "division_by_zero")) 
          return 8;
      } else {
        z = divide(x, y);
        if (ei_x_encode_atom(&result, "ok") || ei_x_encode_double(&result, z)) 
          return 8;
      }
      
    } else {
      if (ei_x_encode_atom(&result, "error") || ei_x_encode_atom(&result, "unsupported_command")) 
        return 99;
    }

    write_cmd(result.buff, result.buffsz);

    ei_x_free(&result);
  }
  return 0;
}

marsharling.cpp

#include <io.h>

typedef unsigned char byte;

int read_exact(byte *buf, int len)
{
  int i, got=0;

  do {
    if ((i = _read(0, buf+got, len-got)) <= 0)
      return(i);
    got += i;
  } while (got<len);

  return(len);
}

int write_exact(byte *buf, int len)
{
  int i, wrote = 0;

  do {
    if ((i = _write(1, buf+wrote, len-wrote)) <= 0)
      return (i);
    wrote += i;
  } while (wrote<len);

  return (len);
}

int read_cmd(byte *buf)
{
  int len;

  if (read_exact(buf, 4) != 4)
    return(-1);
  len = (buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | buf[3];
  return read_exact(buf, len);
}

int write_cmd(byte *buf, int len)
{
  byte li;

  li = (len >> 24) & 0xff;
  write_exact(&li, 1);
  
  li = (len >> 16) & 0xff;
  write_exact(&li, 1);

  li = (len >> 8) & 0xff;
  write_exact(&li, 1);

  li = (len) & 0xff;
  write_exact(&li, 1);

  return write_exact(buf, len);
}

marsharling.h

#ifndef _MARSHARLING_H
#define _MARSHARLING_H

typedef unsigned char byte;

int read_cmd(byte *buf);
int write_cmd(byte *buf, int len);

#endif

port.erl

-module(port).

% API
-export([start/0, start/1, stop/0, add/2, multiply/2, divide/2]).
% Internal exports
-export([init/1]).

start() ->
    start("./Debug/ErlPortTest").
start(ExtPrg) ->
    spawn_link(?MODULE, init, [ExtPrg]).
stop() ->
    ?MODULE ! stop.

add(X, Y) ->
    call_port({add, X, Y}).
multiply(X, Y) ->
    call_port({multiply, X, Y}).
divide(X, Y) ->
    call_port({divide, X, Y}).

call_port(Msg) ->
    ?MODULE ! {call, self(), Msg},
    receive
    Result ->
        Result
    end.

init(ExtPrg) ->
    register(?MODULE, self()),
    process_flag(trap_exit, true),
    Port = open_port({spawn, ExtPrg}, [{packet, 4}, binary, exit_status]),
    loop(Port).

loop(Port) ->
    receive
    {call, Caller, Msg} ->
        io:format("Calling port with ~p~n", [Msg]),
        erlang:port_command(Port, term_to_binary(Msg)),
        receive
        {Port, {data, Data}} ->
            Caller ! binary_to_term(Data);
        {Port, {exit_status, Status}} when Status > 128 ->
            io:format("Port terminated with signal: ~p~n", [Status-128]),
            exit({port_terminated, Status});
        {Port, {exit_status, Status}} ->
            io:format("Port terminated with status: ~p~n", [Status]),
            exit({port_terminated, Status});
        {'EXIT', Port, Reason} ->
            exit(Reason)
        end,
        loop(Port);
    stop ->
        erlang:port_close(Port),
        exit(normal)
    end.