MongoDB Client-Trigger mit Reactive Extensions (Rx)

Ich bin also nicht so der Datenbanktyp, deshalb habe ich mich beim aktuellen Projekt für NoSQL entschieden. Ich denke man findet nur in einem wirklichen Projekt die Schmerzen einer Technologie. Ich will auch nicht in JavaScript programmieren, also ist meine Software in C# implementiert.

Ziel

Eine meiner ersten Anforderungen war, einen clientseitigen Trigger zu bekommen, der von der Datenbank ausgelöst wird, wenn ein Datensatz geändert wird – kann die Datenbank natürlich nicht so. Damit’s später auch schön zu verwenden ist wollte ich es mit den Reactive Extensions implementieren.

So soll’s verwendet werden:

      var elementsChanged = new MongoTrigger(dbClient)
.Where(trigger => trigger.Context == "MyDatabase.Elements");
      elementsChanged.Subscribe(trigger => MessageBox.Show(trigger.ToString()));

Ich habe eine Datenbank “MyDatabase” mit einer Collection “Elements”. Immer wenn Änderungen in dieser Collection erfolgen soll die angegebene Funktion (hier Lambda mit MessageBox) ausgeführt werden.

Der Vorteil ist die Verwendung der Reactive Extensions mit deren Hilfe ganz Einfach eine Filterung über Linq oder die Ausführung auf einem anderen Thread mit ObserveOn möglich.

Recherche

Nach kurzer suche im Internet findet man den Hinweis, dass alle Änderungen der Datenbank bei replizierten Installationen in der Collection oplog gespeichert werden.

Diese Implementierung erfordert, dass die Datenbank als Replica Set konfiguriert ist! (Mehr zu Replikation)

Der nächste Hinweis ist die Existenz von sogenannten tailablen Cursorn. Alles zusammen ergibt eine Abfrage, die bei neuen Einträgen das nächste Ergebnis liefert. Das ist im Prinzip schon die Anforderung für IObservable!

Da die Abfrage natürlich im MoveNext blockiert bis der nächste Eintrag erzeugt wird ist ein Backgroundworker notwendig der das asynchron erledigt.

Implementierung

Damit die Ereignisse alle Änderungsinformationen mitliefern können also zunächst mal eine TriggerArgs Klasse. Damit diese universell eingesetzt werden kann ist die id als string angegeben.

using MongoDB.Bson;
 
namespace ReactiveMongo
{
  
class MongoTriggerArgs
  {
    
public string Context { get; private set
; }
    
public string Operation { get; private set
; }
    
public string Id { get; private set
; }
    
public BsonDocument Document { get; private set
; }
     public MongoTriggerArgs(string context, string operation, string id, BsonDocument document)
     {
       Context = context;
       Operation = operation;
       Id = id;
       Document = document;
     }
    
public override string
ToString()
     {
      
return string.Format("{0}:{1}[{2}]", Operation, Context, Id);
     }
   } }

Als Context wird die Datenbank und die Collection mit Punkt getrennt angegeben (siehe oplog). Operation wird als “u”, “i”, “d” für update, insert, delete angegeben.

In Document steht der gesamte neue Datensatz zur Verfügung.

Hier also die Implementierung:

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics;
using System.Threading;
using MongoDB.Bson;
using MongoDB.Driver;
using MongoDB.Driver.Builders;
 
namespace ReactiveMongo
{
  
class MongoTrigger : IObservable<MongoTriggerArgs>, IDisposable
  {
    
private readonly List<IObserver<MongoTriggerArgs
>> observers;
    
private readonly MongoCollection
opLog;
    
private readonly BackgroundWorker
trigger;
    
public MongoTrigger(MongoClient
client)
     {
       observers =
new List<IObserver<MongoTriggerArgs
>>();
      
var local = client.GetServer().GetDatabase("local"
);
       opLog = local.GetCollection(
"oplog.rs"
);
       trigger =
new BackgroundWorker { WorkerSupportsCancellation = true
};
       trigger.DoWork += TriggerOnDoWork;
       trigger.RunWorkerAsync();
     }
    
private void TriggerOnDoWork(object sender, DoWorkEventArgs
doWorkEventArgs)
     {
var tmax = opLog.FindAllAs<BsonDocument>()
.SetSortOrder(SortBy.Descending(new string[] {"ts"}))
.SetLimit(1)
.First();
var timeStamp = (BsonTimestamp)tmax["ts"];
      
while
(!trigger.CancellationPending)
       {
        
var cursor = opLog.FindAs<BsonDocument>(Query.GTE("ts"
, timeStamp))
                     .SetFlags(
QueryFlags.AwaitData | QueryFlags.TailableCursor | QueryFlags
.NoCursorTimeout)
                     .SetSortOrder(
SortBy.Ascending("$natural"
));
        
using (var enumerator = (MongoCursorEnumerator<BsonDocument
>)cursor.GetEnumerator())
         {
enumerator.MoveNext();
          
while
(!trigger.CancellationPending)
           {
            
if
(enumerator.MoveNext())
             {
              
var
document = enumerator.Current;
              
if (document != null
)
               {
                
Debug
.WriteLine(document.ToString());
                
var context = document["ns"
].AsString;
                
var operation = document["op"
].AsString;
                
var doc = document["o"
].AsBsonDocument;
                
var id = doc["_id"
].AsBsonValue.ToString();
                 NextTrigger(
new MongoTriggerArgs
(context, operation, id, doc));
                 timeStamp = (
BsonTimestamp)document["ts"
];
               }
             }
            
else
            {
              
if
(enumerator.IsDead)
               {
                
break
;
               }
              
if
(!enumerator.IsServerAwaitCapable)
               {
                
Thread.Sleep(TimeSpan
.FromMilliseconds(100));
               }
             }
           }
         }
       }
     }
    
public IDisposable Subscribe(IObserver<MongoTriggerArgs
> observer)
     {
      
if
(!observers.Contains(observer))
         observers.Add(observer);
      
return new Unsubscriber
(observers, observer);
      }
    
private class Unsubscriber : IDisposable
    {
      
private readonly List<IObserver<MongoTriggerArgs
>> unsubscribeObservers;
      
private readonly IObserver<MongoTriggerArgs
> unsubscribeObserver;
      
public Unsubscriber(List<IObserver<MongoTriggerArgs>> observers, IObserver<MongoTriggerArgs
> observer)
       {
         unsubscribeObservers = observers;
         unsubscribeObserver = observer;
       }
      
public void
Dispose()
       {
        
if (unsubscribeObserver != null
&& unsubscribeObservers.Contains(unsubscribeObserver))
           unsubscribeObservers.Remove(unsubscribeObserver);
       }
     }
    
private void NextTrigger(MongoTriggerArgs
args)
     {
      
foreach (var observer in
observers)
       {
         observer.OnNext(args);
       }
     }
    
public void
EndTransmission()
     {
      
foreach (var observer in
observers.ToArray())
        
if
(observers.Contains(observer))
           observer.OnCompleted();
       observers.Clear();
     }
    
public void
Dispose()
     {
      
if (trigger != null)
       {
         trigger.CancelAsync();
       }
     }
   } }

Dies ist die erste Version der Implementierung, in ein paar Stunden erstellt. Ich denke, dass sich beim konkreten Einsatz noch Änderungen und Erweiterungen ergeben.

Link: Daniel Weber: HACK: Creating triggers for MongoDB

Ich bin dankbar für Anregungen…

Advertisements

Über fpf_baden

Ich bin Softwareentwickler aus Leidenschaft.
Dieser Beitrag wurde unter Uncategorized veröffentlicht. Setze ein Lesezeichen auf den Permalink.

Eine Antwort zu MongoDB Client-Trigger mit Reactive Extensions (Rx)

  1. Eduardo schreibt:

    Hello, could you specify the specific versions of dependencies? A complete project with source code would be great. Thank you.

Kommentar verfassen

Trage deine Daten unten ein oder klicke ein Icon um dich einzuloggen:

WordPress.com-Logo

Du kommentierst mit Deinem WordPress.com-Konto. Abmelden / Ändern )

Twitter-Bild

Du kommentierst mit Deinem Twitter-Konto. Abmelden / Ändern )

Facebook-Foto

Du kommentierst mit Deinem Facebook-Konto. Abmelden / Ändern )

Google+ Foto

Du kommentierst mit Deinem Google+-Konto. Abmelden / Ändern )

Verbinde mit %s