method invoke

Source

InvokeResponse invoke(
  Map<String, dynamic> params,
  Responder responder,
  InvokeResponse response,
  Node parentNode,
    [int maxPermission = Permission.CONFIG]) {
  Object rslt;
  try {
    rslt = onInvoke(params);
  } catch (e, stack) {
    var error = new DSError("invokeException", msg: e.toString());
    try {
      error.detail = stack.toString();
    } catch (e) {}
    response.close(error);
    return response;
  }

  var rtype = "values";
  if (configs.containsKey(r"$result")) {
    rtype = configs[r"$result"];
  }

  if (rslt == null) {
    // Create a default result based on the result type
    if (rtype == "values") {
      rslt = {};
    } else if (rtype == "table") {
      rslt = [];
    } else if (rtype == "stream") {
      rslt = [];
    }
  }

  if (rslt is Iterable) {
    response.updateStream(rslt.toList(), streamStatus: StreamStatus.closed);
  } else if (rslt is Map) {
    var columns = [];
    var out = [];
    for (var x in rslt.keys) {
      columns.add({
        "name": x,
        "type": "dynamic"
      });
      out.add(rslt[x]);
    }

    response.updateStream(
      [out],
      columns: columns,
      streamStatus: StreamStatus.closed
    );
  } else if (rslt is SimpleTableResult) {
    response.updateStream(rslt.rows,
        columns: rslt.columns, streamStatus: StreamStatus.closed);
  } else if (rslt is AsyncTableResult) {
    (rslt as AsyncTableResult).write(response);
    response.onClose = (var response) {
      if ((rslt as AsyncTableResult).onClose != null) {
        (rslt as AsyncTableResult).onClose(response);
      }
    };
    return response;
  } else if (rslt is Table) {
    response.updateStream(rslt.rows,
        columns: rslt.columns, streamStatus: StreamStatus.closed);
  } else if (rslt is Stream) {
    var r = new AsyncTableResult();

    response.onClose = (var response) {
      if (r.onClose != null) {
        r.onClose(response);
      }
    };

    Stream stream = rslt;

    if (rtype == "stream") {
      StreamSubscription sub;

      r.onClose = (_) {
        if (sub != null) {
          sub.cancel();
        }
      };

      sub = stream.listen((v) {
        if (v is TableMetadata) {
          r.meta = v.meta;
          return;
        } else if (v is TableColumns) {
          r.columns = v.columns.map((x) => x.getData()).toList();
          return;
        }

        if (v is Iterable) {
          r.update(v.toList(), StreamStatus.open);
        } else if (v is Map) {
          var meta;
          if (v.containsKey("__META__")) {
            meta = v["__META__"];
          }
          r.update([v], StreamStatus.open, meta);
        } else {
          throw new Exception("Unknown Value from Stream");
        }
      }, onDone: () {
        r.close();
      }, onError: (e, stack) {
        var error = new DSError("invokeException", msg: e.toString());
        try {
          error.detail = stack.toString();
        } catch (e) {}
        response.close(error);
      }, cancelOnError: true);
      r.write(response);
      return response;
    } else {
      var list = [];
      StreamSubscription sub;

      r.onClose = (_) {
        if (sub != null) {
          sub.cancel();
        }
      };

      sub = stream.listen((v) {
        if (v is TableMetadata) {
          r.meta = v.meta;
          return;
        } else if (v is TableColumns) {
          r.columns = v.columns.map((x) => x.getData()).toList();
          return;
        }

        if (v is Iterable) {
          list.addAll(v);
        } else if (v is Map) {
          list.add(v);
        } else {
          throw new Exception("Unknown Value from Stream");
        }
      }, onDone: () {
        r.update(list);
        r.close();
      }, onError: (e, stack) {
        var error = new DSError("invokeException", msg: e.toString());
        try {
          error.detail = stack.toString();
        } catch (e) {}
        response.close(error);
      }, cancelOnError: true);
    }
    r.write(response);
    return response;
  } else if (rslt is Future) {
    var r = new AsyncTableResult();

    response.onClose = (var response) {
      if (r.onClose != null) {
        r.onClose(response);
      }
    };

    rslt.then((value) {
      if (value is LiveTable) {
        r = null;
        value.sendTo(response);
      } else if (value is Stream) {
        Stream stream = value;
        StreamSubscription sub;

        r.onClose = (_) {
          if (sub != null) {
            sub.cancel();
          }
        };

        sub = stream.listen((v) {
          if (v is TableMetadata) {
            r.meta = v.meta;
            return;
          } else if (v is TableColumns) {
            r.columns = v.columns.map((x) => x.getData()).toList();
            return;
          }

          if (v is Iterable) {
            r.update(v.toList());
          } else if (v is Map) {
            var meta;
            if (v.containsKey("__META__")) {
              meta = v["__META__"];
            }
            r.update([v], StreamStatus.open, meta);
          } else {
            throw new Exception("Unknown Value from Stream");
          }
        }, onDone: () {
          r.close();
        }, onError: (e, stack) {
          var error = new DSError("invokeException", msg: e.toString());
          try {
            error.detail = stack.toString();
          } catch (e) {}
          response.close(error);
        }, cancelOnError: true);
      } else if (value is Table) {
        Table table = value;
        r.columns = table.columns.map((x) => x.getData()).toList();
        r.update(table.rows, StreamStatus.closed, table.meta);
        r.close();
      } else {
        r.update(value is Iterable ? value.toList() : [value]);
        r.close();
      }
    }).catchError((e, stack) {
      var error = new DSError("invokeException", msg: e.toString());
      try {
        error.detail = stack.toString();
      } catch (e) {}
      response.close(error);
    });
    r.write(response);
    return response;
  } else if (rslt is LiveTable) {
    rslt.sendTo(response);
  } else {
    response.close();
  }

  return response;
}