Rx kann mit einer asynchronen Silverlight Web-Service-Anfrage zu vereinfachen

Question

Ich habe ein vereinfachtes Silverlight-Client-Bibliothek für meine WCF Webservice mit Rx geschrieben, aber merke ich manchmal fehlt mir abgeschlossen Ereignisse.

public IObservable<XElement> GetReport(string reportName)
{
    return from client in Observable.Return(new WebServiceClient())
           from request in Observable.ToAsync<string>(client.GetReportDataAsync)(reportName)
           from result in Observable.FromEvent<GetReportDataCompletedEventArgs>(client, "GetReportDataCompleted").Take(1)
           from close in this.CloseClient(client)
           select result.EventArgs.Result;
}

ich glaube, das Problem durch die Tatsache verursacht wird, dass der Web-Service aufgerufen wird, und kehrt vor dem abgeschlossenen Ereignisse zu abonnieren. Ich kann nicht herausfinden, wie Rx zu bekommen, um das Ereignis zu abonnieren vor dem Async Anruf. Ich habe versucht, StartWith aber das erfordert, dass die Eingangs- und Ausgangstypen gleich sein, irgendwelche Ideen?

Answer 1

Scheint, wie die beste Antwort ist Observable.CreateWithDisposable zu verwenden ()

z.

public IObservable<XElement> GetReport(string reportName)
{
    return from client in Observable.Return(new WebServiceClient())
            from completed in Observable.CreateWithDisposable<GetReportDataCompletedEventArgs>(observer =>
                {
                    var subscription = Observable.FromEvent<GetReportDataCompletedEventArgs>(client, "GetReportDataCompleted")
                        .Take(1)
                        .Select(e => e.EventArgs)
                        .Subscribe(observer);
                    client.GetReportDataAsync(reportName);
                    return subscription;
                })
            from close in this.CloseClient(client)
            select completed.Result;
}

Um diesen leichte Arbeit machen mit mir die CreateWithDisposable in eine gemeinsame Funktion Refactoring, die mit allen Anrufen meines Web-Service verwendet werden können, einschließlich automatisch die Ereignisnamen aus dem Ereignisse args Typ zu bestimmen:

private IObservable<T> CallService<T>(ICommunicationObject serviceClient, Action start) where T : AsyncCompletedEventArgs
{
    if (typeof(T) == typeof(AsyncCompletedEventArgs))
    {
        throw new InvalidOperationException("Event arguments type cannot be used to determine event name, use event name overload instead.");
    }

    string completedEventName = typeof(T).Name.TrimEnd("EventArgs");
    return CallService<T>(serviceClient, start, completedEventName);
}

private IObservable<T> CallService<T>(ICommunicationObject serviceClient, Action start, string completedEventName) where T : AsyncCompletedEventArgs
{
    return Observable.CreateWithDisposable<T>(observer =>
    {
        var subscription = Observable.FromEvent<T>(serviceClient, completedEventName).Take(1).Select(e => e.EventArgs).Subscribe(observer);
        start();
        return subscription;
    });
}

// Example usage:
public IObservable<XElement> GetReport(string reportName)
{
    return from client in Observable.Return(new WebServiceClient())
            from completed in this.CallService<GetReportDataCompletedEventArgs>(client, () => client.GetReportDataAsync(reportName))
            from close in this.CloseClient(client)
            select completed.Result;
}

/// <summary>
/// Asynchronously closes the web service client
/// </summary>
/// <param name="client">The web service client to be closed.</param>
/// <returns>Returns a cold observable sequence of a single success Unit.</returns>
private IObservable<AsyncCompletedEventArgs> CloseClient(WebServiceClient client)
{
    return this.CallService<AsyncCompletedEventArgs>(client, client.CloseAsync, "CloseCompleted");
}

Hope dies hilft jemand anderes!

Answer 2

ich brauche allgemeinen WebClient.DownloadStringAsync zu verwenden, um hier meine Version.

Zuerst wickeln Sie das Ereignis:

public static IObservable<IEvent<DownloadStringCompletedEventArgs>>
    GetDownloadStringObservableEvent(this WebClient wc)
{
    return Observable.FromEvent<DownloadStringCompletedEventArgs>(
        wc, "DownloadStringCompleted");
}

Dann die Erweiterungsmethode erstellen:

public static IObservable<string> GetDownloadString(this WebClient wc, Uri uri)
{
    return Observable.CreateWithDisposable<string>(
        observer => {
            // Several downloads may be going on simultaneously. The token allows
            // us to establish that we're retrieving the right one.
            Guid token = Guid.NewGuid();
            var stringDownloaded = wc.GetDownloadStringObservableEvent()
                    .Where(evt => ((Guid)evt.EventArgs.UserState) == token)
                    .Take(1);        //implicitly unhooks handler after event is received
            bool errorOccurred = false;
            IDisposable unsubscribe =
                stringDownloaded.Subscribe(
                    // OnNext action
                    ev => {
                        // Propagate the exception if one is reported.
                        if (ev.EventArgs.Error != null) {
                            errorOccurred = true;
                            observer.OnError(ev.EventArgs.Error);
                        } else if (!ev.EventArgs.Cancelled) {
                            observer.OnNext(ev.EventArgs.Result);
                        }
                    },
                    // OnError action (propagate exception)
                    ex => observer.OnError(ex),
                    // OnCompleted action
                    () => {
                        if (!errorOccurred) {
                            observer.OnCompleted();
                        }
                    });
            try {
                wc.DownloadStringAsync(uri, token);
            } catch (Exception ex) {
                observer.OnError(ex);
            }
            return unsubscribe;
        }
    );
}

Die Verwendung ist einfach:

wc.GetDownloadString(new Uri("http://myservice"))
    .Subscribe(resultCallback , errorCallback);