Requêtes HttpWeb parallèles avec des extensions réactives

Question

Je dispose d'une classe "Image" avec trois propriétés : Url, Id, Contenu. J'ai une liste de 10 de telles images. Il s'agit d'une application Silverlight.

Je veux créer une méthode :

IObservable TéléchargerImages(List imagesAtelecharger)
{
     //démarrer le téléchargement de toutes les images dans imagesAtelecharger
     //SurTéléchargementImage: 
                          image.Contenu = webResponse.Contenu
                          yield image

}

Cette méthode commence à télécharger toutes les 10 images en parallèle. Ensuite, lorsque le téléchargement de chaque image est terminé, elle définit le Contenu de l'Image sur le Contenu de la WebResponse de ce téléchargement.

Le résultat devrait être un flux IObservable avec chaque image téléchargée.

Je suis débutant en RX, et je pense que ce que je veux peut être réalisé avec ForkJoin, mais c'est dans une version expérimentale du fichier de extensions réactives que je ne veux pas utiliser.

Et je n'aime vraiment pas compter sur les rappels pour détecter que toutes les images ont été téléchargées, puis appeler OnCompleted().

Ça ne me semble pas dans l'esprit de Rx.

Je poste aussi la solution que j'ai codée jusqu'à présent, bien que je n'aime pas ma solution car elle est longue/laid et utilise des compteurs.

     return Observable.Create((IObserver observateur) =>
         {
             int nombreTelechargements = piecesJointesATelecharger.Count;
                foreach (var pieceJointe in piecesJointesATelecharger)
                        {
                            Action action = pieceJointePJ =>
                            this.BeginTelechargerPieceJointe2(pieceJointe).Subscribe(webResponseImage =>
                                {
                                    try
                                    {
                                        using (Stream flux = webResponseImage.GetResponseStream())
                                        {
                                            pieceJointe.ContenuFichier = flux.ReadToEnd();
                                        } 
                                        observateur.OnNext(pieceJointePJ);

                                        lock (verrouNbTelechargements)
                                        {
                                            nombreTelechargements--;
                                            if (nombreTelechargements == 0)
                                            {
                                                observateur.OnCompleted();
                                            }
                                        }
                                    } catch (Exception ex)
                                    {
                                        observateur.OnError(ex);
                                    }
                                });
                            action.Invoke(pieceJointe);
                        }

                        return () => { }; //ne rien faire lorsque l'abonné annule l'abonnement
                    });
            }

D'accord, j'ai réussi à le faire fonctionner à la fin en me basant sur la réponse de Jim.

    var obs = from image in piecesJointesATelecharger.ToObservable()
               from webResponse in this.BeginTelechargerPieceJointe2(image).ObserveOn(TP)
               from responseStream in Observable.Using(webResponse.GetResponseStream, Observable.Return)
               let nouvelleImage = setValeurPieceJointe(image, responseStream.ReadToEnd())
               select nouvelleImage;

où setValeurPieceJointe prend simplement fait `image.Contenu = octets; retour image;

Code de BeginTelechargerPieceJointe2 :

        private IObservable BeginTelechargerPieceJointe2(PieceJointe pieceJointe)
    {
        Uri uriRequete = new Uri(this.LienTelechargementBase + pieceJointe.Id.ToString();
        WebRequest requeteTelechargementImage = HttpWebRequest.Create(uriRequete);
        IObservable observableTelechargementImage = Observable.FromAsyncPattern(requeteTelechargementImage.BeginGetResponse, requeteTelechargementImage.EndGetResponse)();

        return observableTelechargementImage;
    }

Answer 1

Que diriez-vous de simplifier un peu les choses? Prenez votre liste d'images et convertissez-la en un observable. Ensuite, envisagez d'utiliser Observable.FromAsyncPattern pour gérer les demandes de service. Enfin, utilisez SelectMany pour coordonner la demande avec la réponse. Je fais quelques suppositions sur la manière dont vous obtenez les flux de fichiers ici. Fondamentalement, si vous pouvez transmettre les délégués BeginInvoke/EndInvoke dans FromAsyncPattern pour votre demande de service, c'est bon.

var svcObs = Observable.FromAsyncPattern(this.BeginDownloadAttachment2, This.EndDownloadAttchment2);

var obs = from image in imagesToDownload.ToObservable()
          from responseStream in svcObs(image)
          .ObserveOnDispatcher()
          .Do(response => image.FileContent = response.ReadToEnd())
          select image;
return obs;