Transform IObservable of strings into IObservable of XDocuments

I have an IObservable<string> that contains (fragments of) XML documents. I'd like to transform one into the other. So for example, suppose I have the following fragments that are pushed from my IObservable<string> (each line contains a fragment):

<?xml version=
"1.0" ?>
<t开发者_运维技巧estXml></test
Xml><?xml version="1.0"
?><otherXml /><?xm

How can I morph this into an IObservable<XDocument> to get the following documents:

<?xml version="1.0"?><testXml />
<?xml version="1.0"?><otherXml />

I've been thinking about handing the IObservable<string> to some blocking TextReader implementation but I think there should be a more clever solution.

How about this:

IObservable<string> splitXmlTokensIntoSeparateLines(string s)
{
    // Here, you need to split tokens into separate lines (where 'token'
    // is the beginning of an Xml element). This makes it easier down
    // the line for the TakeWhile operator.

    return new[] { firstPart, secondPart, etc }.ToObservable();
}

bool doesTokenTerminateDocument(string s)
{
    // Here, you should return whether the XML represents the end of one 
    // document
}

var xmlDocuments = stringObservable
    .SelectMany(x => splitXmlTokensIntoSeparateLines(x))
    .TakeWhile(x => doesTokenTerminateDocument(x))
    .Aggregate(new StringBuilder(), (acc, x), acc.Append(x))
    .Select(x => {
        var ret = new XDocument();
        ret.Parse(x.ToString());
        return ret;
    })
    .Repeat()
    .TakeUntil(stringObservable.Aggregate(0, (acc, _) => acc));

The TakeUntil is a hack to make it terminate correctly - basically, Repeat would keep resubscribing forever, unless we stop it by telling it to finish when stringObservable completes.

Any reason why you can't use the Select operator?

var xmlObs = stringObs.Select(s => {
  var x = new XDocument();
  x.Parse(s);
  return x;
});