lprun doesn't respect output format with reactive sources

raboof

When dumping an IObservable<T> source, lprun doesn't respect the requested output format via the -format argument. Instead, it dumps a JSON rendering of each individual object and where the output stream as a whole is invalid JSON. For example, save the following (expression) query in a file called test.linq:

from x in Observable.Range(1, 10)
select new { X = x, SquareOfX = x * x }

Then run with lprun -format=csv test.linq. The output will be as follows:

{
  "X": 1,
  "SquareOfX": 1
}{
  "X": 2,
  "SquareOfX": 4
}{
  "X": 3,
  "SquareOfX": 9
}{
  "X": 4,
  "SquareOfX": 16
}{
  "X": 5,
  "SquareOfX": 25
}{
  "X": 6,
  "SquareOfX": 36
}{
  "X": 7,
  "SquareOfX": 49
}{
  "X": 8,
  "SquareOfX": 64
}{
  "X": 9,
  "SquareOfX": 81
}{
  "X": 10,
  "SquareOfX": 100
}

What's more, lprun doesn't fail with a non-zero exit code if the source reports an error. To see this in action, change `test.linq` content to the following:

from x in Observable.Range(-5, 10)
select new { X = x, Y = 100 / x }

This time, run with lprun -format=csv test.linq && echo OK. The output will be:

{
  "X": -5,
  "Y": -20
}{
  "X": -4,
  "Y": -25
}{
  "X": -3,
  "Y": -33
}{
  "X": -2,
  "Y": -50
}{
  "X": -1,
  "Y": -100
}{
  "Message": "Attempted to divide by zero.",
  "Data": [],
  "InnerException": null,
  "TargetSite": {
    "Name": "b__0_0",
    "DeclaringType": "typeof(<>c)",
    "ReflectedType": "typeof(<>c)",
    "CustomAttributes": [],
    "MetadataToken": 100663307,
    "MethodImplementationFlags": 0,
    "MethodHandle": {
      "Value": {}
    },
    "Attributes": 131,
    "CallingConvention": 33,
    "IsGenericMethodDefinition": false,
    "ContainsGenericParameters": false,
    "IsGenericMethod": false,
    "IsSecurityCritical": true,
    "IsSecuritySafeCritical": false,
    "IsSecurityTransparent": false,
    "IsPublic": false,
    "IsPrivate": false,
    "IsFamily": false,
    "IsAssembly": true,
    "IsFamilyAndAssembly": false,
    "IsFamilyOrAssembly": false,
    "IsStatic": false,
    "IsFinal": false,
    "IsVirtual": false,
    "IsHideBySig": true,
    "IsAbstract": false,
    "IsSpecialName": false,
    "IsConstructor": false,
    "MemberType": 8,
    "ReturnType": "typeof(o)",
    "ReturnParameter": {
      "ParameterType": "typeof(o)",
      "Name": null,
      "HasDefaultValue": true,
      "DefaultValue": null,
      "RawDefaultValue": null,
      "Position": -1,
      "Attributes": 0,
      "Member": {},
      "IsIn": false,
      "IsOut": false,
      "IsLcid": false,
      "IsRetval": false,
      "IsOptional": false,
      "MetadataToken": 134217728,
      "CustomAttributes": []
    },
    "ReturnTypeCustomAttributes": {
      "ParameterType": "typeof(o)",
      "Name": null,
      "HasDefaultValue": true,
      "DefaultValue": null,
      "RawDefaultValue": null,
      "Position": -1,
      "Attributes": 0,
      "Member": {},
      "IsIn": false,
      "IsOut": false,
      "IsLcid": false,
      "IsRetval": false,
      "IsOptional": false,
      "MetadataToken": 134217728,
      "CustomAttributes": []
    }
  },
  "StackTrace": "   at UserQuery.<>c.b__0_0(Int32 x)\r\n   at System.Reactive.Linq.ObservableImpl.Select`2._.OnNext(TSource value)",
  "HelpLink": null,
  "Source": "LINQPadQuery",
  "HResult": -2147352558
}

More importantly, echo OK will execute and print OK because lprun terminates with an exit code of zero.