internal static IErlObject DecodeRPC(IErlObject msg)
        {
            var term = msg as ErlTuple;
              if (term == null)
            return null;

              var binding = term.Match(RpcReplyPattern);
              return binding != null ? binding[T] : null;
        }

    /// <summary>
    /// Create a stream containing the serialized Erlang term.
    /// Optionally include in the beginning Erlang protocol version byte
    /// </summary>
    public ErlOutputStream(IErlObject o = null, bool writeVersion = true, bool writePktSize = false,
        int capacity = DEFAULT_INITIAL_SIZE)
    {
      m_Buffer = new byte[capacity];
      m_Capacity = capacity;
      m_Position = 0;

      if (writePktSize)
        Write4BE(0); // make space for length data, but final value is not yet known

      if (o == null)
      {
        if (writeVersion)
          write(ErlExternalTag.Version);
        return;
      }

      encodeObject(o, writeVersion);

      if (writePktSize)
        Poke4BE(0, m_Position - 4);
    }

		    /// <summary>
		    /// Create an Erlang tuple from the given list of items
		    /// </summary>
		    public ErlTuple(IErlObject[] items) : base(items) {}

 /// <summary>
 /// Write an arbitrary Erlang term to the stream.
 /// </summary>
 public void Write(IErlObject o)
 {
   switch (o.TypeOrder)
   {
     case ErlTypeOrder.ErlAtom:      WriteAtom((ErlAtom)o); break;
     case ErlTypeOrder.ErlBinary:    WriteBinary((ErlBinary)o); break;
     case ErlTypeOrder.ErlBoolean:   WriteBoolean((ErlBoolean)o); break;
     case ErlTypeOrder.ErlByte:      WriteByte((ErlByte)o); break;
     case ErlTypeOrder.ErlDouble:    WriteDouble((ErlDouble)o); break;
     case ErlTypeOrder.ErlLong:      WriteLong((ErlLong)o); break;
     case ErlTypeOrder.ErlList:      WriteList((ErlList)o); break;
     case ErlTypeOrder.ErlPid:       WritePid((ErlPid)o); break;
     case ErlTypeOrder.ErlPort:      WritePort((ErlPort)o); break;
     case ErlTypeOrder.ErlRef:       WriteRef((ErlRef)o); break;
     case ErlTypeOrder.ErlString:    WriteString((ErlString)o); break;
     case ErlTypeOrder.ErlTuple:     WriteTuple((ErlTuple)o); break;
     default:
       throw new ErlException(
           StringConsts.ERL_UNSUPPORTED_ELEMENT_TYPE_ERROR,
               o.GetType().Name, o.TypeOrder.ToString());
   }
 }

    /// <summary>
    /// Used to break all known links to this mbox
    /// </summary>
    internal void BreakLinks(ErlAtom fromNode, IErlObject reason)
    {
      var links = m_Links.Remove(fromNode);

      foreach (var link in links)
        if (link.HasPid)
          m_Node.Deliver(ErlMsg.Exit(m_Self, link.Pid, reason));
        else
          m_Node.Deliver(new ErlConnectionException(fromNode, reason));

      foreach (var m in m_Monitors.Where(o => o.Value.Node == fromNode)
                                  .Where(m => m_Monitors.Remove(m.Key)))
        Deliver(new ErlConnectionException(fromNode, reason));
    }

 /// <summary>
 /// Send a message to a named mailbox created from another node
 /// </summary>
 public bool Send(ErlAtom node, ErlAtom name, IErlObject msg)
 {
   return m_Node.Deliver(node, ErlMsg.RegSend(m_Self, name, msg));
 }

 /*
 * send to remote name
 * dest is recipient's registered name, the nodename is implied by
 * the choice of connection.
 */
 public void Send(ErlPid from, ErlAtom dest, IErlObject msg)
 {
   // encode and send the message
   base.Send(ErlMsg.RegSend(from, dest, msg, SendCookie));
 }

 internal static ErlTuple EncodeRPC(
     ErlPid from, ErlAtom mod, ErlAtom fun, ErlList args, IErlObject gleader)
 {
   /*{Self, {call, Mod, Fun, Args, GroupLeader}} */
   return new ErlTuple(from, new ErlTuple(ConstAtoms.Call, mod, fun, args, gleader));
 }

 internal void IoOutput(ErlAtom encoding, IErlObject output)
 {
     if (m_OnIoOutput != null)
         m_OnIoOutput(encoding, output);
 }

 /// <summary>
 /// Break links of all pids linked to pids on the fromNode node
 /// </summary>
 internal void BreakLinks(ErlAtom fromNode, IErlObject reason)
 {
     foreach (var m in m_Mailboxes)
         m.Value.BreakLinks(fromNode, reason);
 }

 /// <summary>
 /// Send a message to a named mailbox on local node
 /// </summary>
 public bool Send(ErlPid from, ErlAtom toName, IErlObject msg)
 {
     return Deliver(ErlMsg.RegSend(from, toName, msg));
 }

 /// <summary>
 /// Create an Erlang string from the given string
 /// </summary>
 public ErlList(IErlObject[] items, bool clone = true)
     : base(items, clone)
 {
 }

 /// <summary>
 /// Determine if two Erlang objects are equal
 /// </summary>
 public override bool Equals(IErlObject o)
 {
     if (o is ErlList) return Equals((ErlList)o);
     if (o is ErlString)
     {
       var rhs = ((ErlString)o).Value;
       if (rhs.Length != m_Items.Count) return false;
       return !rhs.Where((t, i) => !m_Items[i].IsInt() || (int)t != m_Items[i].ValueAsInt).Any();
     }
     return false;
 }

 /// <summary>
 /// Determine if two Erlang objects are equal
 /// </summary>
 public override bool Equals(IErlObject o)
 {
     return o is ErlTuple ? Equals((ErlTuple)o) : false;
 }

    private IErlObject rpcCall(ErlPid from, ErlRef eref,
        ErlAtom mod, ErlAtom fun, ErlList args, IErlObject groupLeader)
    {
      // We spawn a new task, so that RPC calls wouldn't block the RPC server thread
      Task.Factory.StartNew(() =>
      {
        var type = Type.GetType(mod);

        if (type == null)
        {
          sendRpcReply(from, eref,
              ErlTuple.Create(ConstAtoms.Error, "unknown type: {0}".Args(mod)));
          return;
        }

        // TODO: add LRU caching
        //var method = type.GetMethod(fun.Value, BindingFlags.Static | BindingFlags.Public);

        string methodName = fun.Value;

        if (args.Count == 0)
        {
          var pi = type.GetProperty(fun, BindingFlags.Static | BindingFlags.Public);
          if (pi != null)
          {
            try
            {
              var result = pi.GetValue(null, null);
              sendRpcReply(from, eref,
                  ErlTuple.Create(ConstAtoms.Ok, result.ToErlObject()));
            }
            catch (Exception e)
            {
              sendRpcReply(from, eref,
                  ErlTuple.Create(ConstAtoms.Error, new ErlString(e.Message)));
            };
            return;
          }
        }

        var mi = type.GetMethods(BindingFlags.Static | BindingFlags.Public)
                     .Where(m => m.Name == methodName && m.GetParameters().Count() == args.Count)
                     .FirstOrDefault();

        if (mi == null)
        {
          sendRpcReply(from, eref,
              ErlTuple.Create(ConstAtoms.Error, "unknown method: {0}".Args(fun)));
          return;
        }

        var pars = mi.GetParameters();

        var margs = new object[pars.Length];

        for (int i = 0; i < pars.Length; i++)
        {
          var par = pars[i];
          var val = args[i];
          margs[i] = val.AsType(par.ParameterType);
        }

        try
        {
          var result = mi.Invoke(type, margs);
          sendRpcReply(from, eref, ErlTuple.Create(ConstAtoms.Ok, result.ToErlObject()));
        }
        catch (Exception e)
        {
          sendRpcReply(from, eref, ErlTuple.Create(ConstAtoms.Error, new ErlString(e.Message)));
        }
      });
      return (IErlObject)null;
    }

 internal static ErlTuple EncodeRPCcast(
     ErlPid from, string mod, string fun, ErlList args, IErlObject gleader)
 {
   return EncodeRPCcast(from, new ErlAtom(mod), new ErlAtom(fun), args, gleader);
 }

        private void checkForError(IErlObject pattern, IErlObject response, ErlVarBind bind, int reqID)
        {
            if (bind != null)
              return;

            bind = response.Match(pattern);
            if (bind == null)
              throw new ErlDataAccessException(StringConsts.ERL_DS_INVALID_RESP_PROTOCOL_ERROR + "invalid error response pattern", new Exception(response.ToString()));

            var gotReqID = bind[ATOM_ReqID].ValueAsLong;

            if (gotReqID != reqID)
              throw new ErlDataAccessException(StringConsts.ERL_DS_INVALID_RESP_PROTOCOL_ERROR + "unexpected transaction ID (expected={0}, got={1})".Args(reqID, gotReqID));

            var ecode = bind[ATOM_Code].ValueAsInt;
            var rmsg  = bind[ATOM_Msg];
            var emsg  = rmsg.TypeOrder == ErlTypeOrder.ErlString || rmsg.TypeOrder == ErlTypeOrder.ErlAtom
                  ? rmsg.ValueAsString
                  : rmsg.ToString();

            Exception error;

            switch (ecode)
            {
              case INVALID_SUBSCRIPTION_REQUEST_EXCEPTION:
            error = new NFX.DataAccess.CRUD.Subscriptions.InvalidSubscriptionRequestException(emsg, null);
            break;
              case -1:
            error = new ErlDataAccessException("Remote error message: {0}".Args(emsg));
            break;
              default:
            error = new ErlDataAccessException("Remote error code {0}. Message: {1}".Args(ecode, emsg));
            break;
            }

            throw error;
        }

 private IErlObject ioProcessGetLine(
     ErlAtom encoding, IErlObject prompt, IErlObject replyAs)
 {
   return s_ReplyPattern.Subst(
       new ErlVarBind{
                     {RA, replyAs},
                     {R, (IErlObject)Tuple.Create(ConstAtoms.Error, ConstAtoms.Request)}
       });
 }

    /// <summary>
    /// Receive a message and match it against a given pattern
    /// </summary>
    /// <param name="pattern">Pattern to match the message against</param>
    /// <param name="timeoutMsec">Timeout in milliseconds</param>
    /// <returns>Return a tuple containing the received message and variable
    /// binding object. On timeout the first element of the tuple is null.
    /// On unsuccessful match the second element of the tuple is null</returns>
    public Tuple<IErlObject, ErlVarBind> ReceiveMatch(IErlObject pattern, int timeoutMsec = -1)
    {
      var m = Receive(timeoutMsec);
      if (m == null) return Tuple.Create<IErlObject, ErlVarBind>(null, null);

      var binding = new ErlVarBind();
      bool res = m.Match(pattern, binding);
      return Tuple.Create(m, res ? binding : null);
    }

    private IErlObject ioProcessRequests(
        ErlPatternMatcher pm, ErlList requests, IErlObject replyAs)
    {

      foreach (var r in requests)
      {
        IErlObject term = r;
        if (pm.Match(ref term, replyAs) < 0)
          return term;
      }
      return s_ReplyPattern.Subst(
          new ErlVarBind{
                        {RA, replyAs},
                        {R, (IErlObject)Tuple.Create(ConstAtoms.Error, ConstAtoms.Request)
                    }});
    }