Windows Phone 7 + Reactive ExtensionsによるXml取得

public MainPage()
{
    InitializeComponent();
 
    var wc = new WebClient();
    wc.OpenReadCompleted += (sender, e) =>
    {
        var elem = XElement.Load(e.Result); // e.ResultにStreamが入ってる
        MessageBox.Show(elem.ToString()); // 確認
    };
    wc.OpenReadAsync(new Uri("http://twitter.com/statuses/public_timeline.xml")); // 非同期読み込み呼び出し開始
}

var wc = new WebClient();
wc.AllowReadStreamBuffering = false; // デフォはtrueでバッファリングあり、今回はfalseに変更
wc.OpenReadCompleted += (sender, e) =>
{
    try
    {
        var elem = XElement.Load(e.Result); // ここで例外出るよ！
    }
    catch (Exception ex)
    {
        // Read is not supporeted on the main thread when buffering is disabled.
        MessageBox.Show(ex.ToString());
    }
};

wc.OpenReadCompleted += (sender, e) =>
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        try
        {
            var elem = XElement.Load(e.Result);
            MessageBox.Show(elem.ToString()); // 今度はここで例外！
        }
        catch(Exception ex)
        {
            // Invalid cross-thread access.
            Debug.WriteLine(ex.ToString());
        }
    });
};

wc.OpenReadCompleted += (sender, e) =>
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        var elem = XElement.Load(e.Result);
        Dispatcher.BeginInvoke(() => MessageBox.Show(elem.ToString()));
    });
};

var wc = new WebClient { AllowReadStreamBuffering = false };
 
Observable.FromEvent<OpenReadCompletedEventArgs>(wc, "OpenReadCompleted")
    .ObserveOn(Scheduler.ThreadPool) // ThreadPoolで動かすようにする
    .Select(e => XElement.Load(e.EventArgs.Result))
    .ObserveOnDispatcher() // UIスレッドに戻す
    .Subscribe(x => MessageBox.Show(x.ToString()));
 
wc.OpenReadAsync(new Uri("http://twitter.com/statuses/public_timeline.xml"));

var req = WebRequest.CreateHttp("http://twitter.com/statuses/public_timeline.xml");
req.AllowReadStreamBuffering = false;
req.BeginGetResponse(ar =>
{
    using (var res = req.EndGetResponse(ar))
    using (var stream = res.GetResponseStream())
    {
        var x = XElement.Load(res.GetResponseStream());
        Dispatcher.BeginInvoke(() => MessageBox.Show(x.ToString()));
    }
}, null);

var req = HttpWebRequest.CreateHttp("http://twitter.com/statuses/public_timeline.xml");
req.AllowReadStreamBuffering = false;
 
Observable.FromAsyncPattern<WebResponse>(req.BeginGetResponse, req.EndGetResponse)
    .Invoke() // 非同期実行開始(Invoke()じゃなくて()でもOKです、ただのDelegateなので)
    .Select(res => XElement.Load(res.GetResponseStream()))
    .ObserveOnDispatcher()
    .Subscribe(x => MessageBox.Show(x.ToString()));

<Grid x:Name="LayoutRoot" Background="Transparent">
    <Grid.RowDefinitions>
        <RowDefinition Height="Auto"/>
        <RowDefinition Height="*"/>
    </Grid.RowDefinitions>
 
    <Button Grid.Row="0" Height="72" Width="200" Content="Reload" Name="Reload"></Button>
    <ListBox Grid.Row="1" Name="TweetList" DataContext="{Binding}">
        <ListBox.ItemTemplate>
            <DataTemplate>
                <StackPanel Orientation="Horizontal">
                    <Image Source="{Binding Image}" Height="73" Width="73" VerticalAlignment="Top" />
                    <StackPanel Width="350">
                        <TextBlock Text="{Binding Name}" Foreground="Red" />
                        <TextBlock Text="{Binding Text}" TextWrapping="Wrap" />
                    </StackPanel>
                </StackPanel>
            </DataTemplate>
        </ListBox.ItemTemplate>
    </ListBox>
</Grid>

public class TwitterStatus
{
    public long Id { get; set; }
    public string Text { get; set; }
    public string Name { get; set; }
    public string Image { get; set; }
 
    public TwitterStatus(XElement element)
    {
        Id = (long)element.Element("id");
        Text = (string)element.Element("text");
        Name = (string)element.Element("user").Element("screen_name");
        Image = (string)element.Element("user").Element("profile_image_url");
    }
}
 
public partial class MainPage : PhoneApplicationPage
{
    public MainPage()
    {
        InitializeComponent();
        Reload.Click += new RoutedEventHandler(Reload_Click); // XAMLに書いてもいいんですけど。
    }
 
    void Reload_Click(object sender, RoutedEventArgs e)
    {
        var req = HttpWebRequest.CreateHttp("http://twitter.com/statuses/public_timeline.xml");
        req.AllowReadStreamBuffering = false;
 
        Observable.FromAsyncPattern<WebResponse>(req.BeginGetResponse, req.EndGetResponse)
            .Invoke()
            .Select(res => XElement.Load(res.GetResponseStream()))
            .Select(x => x.Descendants("status").Select(xe => new TwitterStatus(xe)))
            .ObserveOnDispatcher()
            .Subscribe(ts => TweetList.ItemsSource = ts);
    }
}

// 30秒毎もしくはリロードボタンクリックでPublicTimeLineを更新
Observable.Merge(
        Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(30), Scheduler.NewThread).Select(_ => (object)_),
        Observable.FromEvent<RoutedEventArgs>(Reload, "Click").Select(_ => (object)_))
    .SelectMany(_ =>
    {
        var req = HttpWebRequest.CreateHttp("http://twitter.com/statuses/public_timeline.xml");
        req.AllowReadStreamBuffering = false;
        return Observable.FromAsyncPattern<WebResponse>(req.BeginGetResponse, req.EndGetResponse)();
    })
    .Select(res => XStreamingReader.Load(res.GetResponseStream()))
    .SelectMany(x => x
        .Descendants("status")
        .Select(xe => new TwitterStatus(xe))
        .Reverse()) // 古い順にする
    .Scan((before, current) => before.Id > current.Id ? before : current) // 最後に通した記事よりも古ければ通さない(で、同じ記事を返す)
    .DistinctUntilChanged(t => t.Id) // 同じ記事が連続して来た場合は何もしないでreturn
    .ObserveOnDispatcher()
    .Subscribe(t => TweetList.Items.Insert(0, t)); // Insertだって...