Correct pattern for chain of transformations (pipeline) on observable stream?

I have the following scenario:

• taking into account the flow of objects IObservable<E>
  
  
• handle each E
  
  one to get either an E1
  
  error state, in which case I need an error messageM1
  
  
• process each E1
  
  one to get either E2
  
  an error messageM2
  
  
• ....

There is an additional complication is that the result of En and / or an error message Mn

may depend on the values of E

, E1

, ..., En-1

- not only from En-1

.

With all of this in mind, is there a better sample than what I am using?

[Edit] As requested, I've added a fully working example; unfortunately made this post quite large.

internal class Program
{
  private static void Main()
  {
    var stream = Enumerable.Range(1, 10).Select(i => new Record { Id = i }).ToObservable();

    stream
      .Select(it => new ComplexType { Item = it })
      .SelectIfOk(Process1)
      .SelectIfOk(Process2)
      .SelectIfOk(ProcessN)
      .Subscribe(DisplayResult);

    Console.ReadLine();
  }

  private static ComplexType Process1(ComplexType data)
  {
    // do some processing
    data.E1 = data.Item.Id * 10;

    // check for errors in output
    if (data.E1 == 30 || data.E1 == 70)
    {
      data.Errors.Add("Error");
    }

    return data;
  }

  private static ComplexType Process2(ComplexType data)
  {
    // do some processing
    data.E2 = (data.E1 - 3).ToString();

    // check for errors in output
    // can generate multiple errors for the same item
    if (data.E2.StartsWith("4"))
    {
      // does not only depend on the immediate precursor, E1 in this case
      data.Errors.Add("Starts with 4 -- " + data.Item.Id);
    }

    if (data.E2.StartsWith("8"))
    {
      data.Errors.Add("Starts with 8");
    }

    return data;
  }

  private static ComplexType ProcessN(ComplexType data)
  {
    // do some processing
    data.EN = "Success " + data.E2;

    // this one doesn't generate errors
    return data;
  }

  private static void DisplayResult(ComplexType data)
  {
    if (data.Errors.Any())
    {
      Console.WriteLine("{0:##0} has errors: " + string.Join(",", data.Errors));
    }
    else
    {
      Console.WriteLine("{0:##0}: {1}", data.Item.Id, data.EN);
    }
  }
}

      

        
        
        
      

    

These are the classes used in the above code example:

public class Record
{
  public int Id { get; set; }
  public string FullName { get; set; }
  public string OtherStuff { get; set; }
}

public class ComplexType
{
  public Record Item { get; set; }

  // intermediary results
  public int E1 { get; set; }
  public string E2 { get; set; }

  // final result
  public string EN { get; set; }

  public List<string> Errors { get; set; }

  public ComplexType()
  {
    Errors = new List<string>();
  }
}

      

        
        
        
      

    

Note that between the types E1

, E2

..., En

no links between them (in particular, they did not all inherit the same general type).

SelectIfOk

is an extension method:

public static IObservable<T> SelectIfOk<T>(this IObservable<T> observable,
  Func<T, T> selector)
  where T : ComplexType
{
  return observable.Select(item => item.Errors.Any() ? item : selector(item));
}

      

        
        
        
      

    

The result of running this code is:

1: Success 7
2: Success 17
3 has errors: Error
4: Success 37
5 has errors: Starts with 4 -- 5
6: Success 57
7 has errors: Error
8: Success 77
9 has errors: Starts with 8
10: Success 97

      

        
        
        
      

    

I am using ComplexType so that I can carry both intermediate results and error state and it just looks like ... fishy. I've been looking at this code for a week (it was a hobby project) and I continue to feel like I'm missing the right way to do things with Rx.

[Edit] I forgot to mention a very important thing: it is very important that I process all elements in the stream, even if some of them generate errors; so I can't just use an overload Subscribe

that takes an exception - it will terminate the thread. Abandoning one item when the error is good (if Process1

generates an error, then Process2

, ..., are ProcessN

no longer executed), but does not leave the whole thread.

[Edit] One more clarification: if it helps, the processing I mean is more natural for the TPL DataFlow library, except that I'm limited to .NET 4.0 so I can't use it.

BTW, I couldn't find any serious discussion of error handling in Rx, it usually mentions the Subscribe

overload / the call OnError

and what it is. Does anyone have any guidelines for an in-depth approach to the topic?