RX - Erreur distincte avec la personnalisation IEqualityComparer

Question

J'ai un protocole basé sur l'interrogation et je veux profiter de RX pour le transformer en un push. Toutes les x secondes je demande des Tags (nom et valeur) en utilisant le protocole, et je les récupère.

Je ne suis intéressé que par le changement de valeurs dans les balises, et j'utilise donc la fonction DistinctUntilChanges.

this.TagsChangeNotifier = _tags 
    .Select(tag => 
    { 
     return Observable 
      .Interval(ts) 
      .Select(_ => { return tag; }) 
      .DistinctUntilChanged(new DataTagComparer()); 
    }) 
    .Merge(); 

Et voici la classe DataTagcomparer.

public class DataTagComparer : IEqualityComparer<DataTag> 
{ 

    public bool Equals(DataTag x, DataTag y) 
    { 
      b = y.WeakRawValue.ToByteArray().SequenceEqual(x.WeakRawValue.ToByteArray()); 

     return b; 
    } 

    public int GetHashCode(DataTag obj) 
    { 
     return obj.Name.GetHashCode(); 
    } 
} 

Mais ça ne marche pas, parce que je ne vois jamais de comparaison entre 2 valeurs différentes. Voici un exemple.

Start program: DataTag("Test",1) 
Equals called: x = ("Test",1), y = ("Test",1) 

Attendre 10 secondes et le changement de protocole pour revenir 2 au lieu de 1.

Equals called: x = ("Test",1), y = ("Test",1) 
Equals called: x = ("Test",2), y = ("Test",2) 
Equals called: x = ("Test",2), y = ("Test",2) 
Equals called: x = ("Test",2), y = ("Test",2) 

et ainsi de suite.

La partie étrange est qu'il manque complètement la comparaison entre la valeur précédente et la valeur actuelle! Savez-vous quel peut être le problème? En fait, je travaille avec cette solution de contournement terrible

public class DataTagComparer : IEqualityComparer<DataTag> 
{ 

    private object val; 

    public bool Equals(DataTag x, DataTag y) 
    { 
     bool b = true; 

     if (val != null) 
      b = val.ToByteArray().SequenceEqual(x.WeakRawValue.ToByteArray()); 

     val = x.WeakRawValue; 

     return b; 
    } 

    public int GetHashCode(DataTag obj) 
    { 
     return obj.Name.GetHashCode(); 
    } 
} 

Merci pour votre attention, Vincenzo.

EDIT: DataTag code de classe

public abstract class DataTag 
{ 
    public DataTag(string _Name, string Desc) 
    { 
     Name = _Name; 
     Description = Desc; 
    } 
    public string Name { get; private set; } 
    public string Description { get; private set; } 
    public abstract object WeakValue { get; } 
    public abstract object WeakRawValue { get; } 
} 

EDIT: la fonction de mise à jour Tag

this.timerHandle = Observable.Interval(ts).Select(_ => { Update(); return _; }).Publish().Connect(); 

Answer 1

Ce .... ne semble pas juste, en fonction de votre description - bien que je pourrais être vous interprète mal ...

this.TagsChangeNotifier = _tags 
    // for each tag value in tags... 
    .Select(tag => 
    { 
     // Tick off every TimeSpan ts, then... 
     return Observable.Interval(ts) 
      // Say we've "ticked" 
      .Do(tick => Console.WriteLine("It's time to tick!")) 
      // Return the value "tag" (which remains constant...) 
      .Select(_ => { return tag; }) 
      // Say what we see 
      .Do(t => Console.WriteLine("I see a {0}!", t)) 
      // But only when it's different from the last one 
      // (but we never change the value?) 
      .DistinctUntilChanged(new DataTagComparer()); 
    })  
    // And mash them all together into one stream 
    .Merge(); 

Je suppose que tout dépend de ce que _tags est et à certains de gree quelle est la définition de DataTag, mais je ne pense pas que ce soit ce que vous voulez vraiment.

EDIT:

Tirons le flux - en commençant par _tags, qui pour l'instant je suppose est un IObservable:

Time _tags 
    | tag1 
    | tag2 
    | tag3 

Jusqu'à présent, si bon - maintenant pour chacun de ces , nous créons un Select et Interval:

Time _tags 
    | tag1 
    |  \---- Interval 
    | 
    | tag2 
    |  \---- Interval 
    | 
    | tag3 
    |  \---- Interval 
    | 

et nous tick pendant un certain temps, sélectionner à nouveau l'étiquette à chaque fois:

Time _tags 
    | tag1 
    |  \---- Interval 
    |   \-------Tick -> tag1 
    |   \-------Tick -> tag1 
    |   \-------Tick -> tag1 
    | tag2 
    |  \---- Interval 
    |   \-------Tick -> tag2 
    |   \-------Tick -> tag2 
    |   \-------Tick -> tag2 
    | tag3 
    |  \---- Interval 
    |   \-------Tick -> tag3 
    |   \-------Tick -> tag3 
    |   \-------Tick -> tag3 
    | 

Ensuite, nous ajoutons le DistinctUntilChanged:

Time _tags 
    | tag1 
    |  \---- Interval 
    |   \-------Tick -> tag1 ---> tag1 
    |   \-------Tick -> tag1 -X 
    |   \-------Tick -> tag1 -X 
    | tag2 
    |  \---- Interval 
    |   \-------Tick -> tag2 ---> tag2 
    |   \-------Tick -> tag2 -X 
    |   \-------Tick -> tag2 -X 
    | tag3 
    |  \---- Interval 
    |   \-------Tick -> tag3 ---> tag3 
    |   \-------Tick -> tag3 -X 
    |   \-------Tick -> tag3 -X 
    | 

Et enfin, nous Merge les flux de sous:

Time _tags          Output 
    | tag1           | 
    |  \---- Interval        | 
    |   \-------Tick -> tag1 ---> tag1  tag1 
    |   \-------Tick -> tag1 -X    | 
    |   \-------Tick -> tag1 -X    | 
    | tag2           | 
    |  \---- Interval        | 
    |   \-------Tick -> tag2 ---> tag2  tag2 
    |   \-------Tick -> tag2 -X    | 
    |   \-------Tick -> tag2 -X    | 
    | tag3           | 
    |  \---- Interval        | 
    |   \-------Tick -> tag3 ---> tag3  tag3 
    |   \-------Tick -> tag3 -X    | 
    |   \-------Tick -> tag3 -X    | 
    | 

Donc, si tout ce que vous devez faire est prise lorsque le flux de valeurs changements, vous pouvez essayer quelque chose de la forme suivante:

// my fake source of "tags", in this case simple strings 
var subject = new Subject<string>(); 
var source = subject.Publish().RefCount(); 

// Still want to track "distinct chains" 
var distincts = source.DistinctUntilChanged(); 
// But we also want to "look into the future", and see the *next* distinct chain 
var futureDistincts = source.DistinctUntilChanged().Skip(1); 
// A "delta" occurs when a distinct chain ends, so we'll zip the two 
// sequences together (since they are "now distinct" and "now + 1", this will mean changes) 
var onlyDeltas = distincts 
    .Zip(futureDistincts, (before,after) => Tuple.Create(before,after)); 

using(onlyDeltas.Subscribe(Console.WriteLine)) 
{ 
    subject.OnNext("Foo"); 
    subject.OnNext("Foo"); 
    subject.OnNext("Foo"); 
    subject.OnNext("Bar"); // BAM: triggers an output value of (Foo, Bar) 
    subject.OnNext("Bar"); 
    subject.OnNext("Foo"); // BAM: triggers an output value of (Bar, Foo) 
}