228.11. Oplog テールトラッキング

oplog コレクション追跡機能により、MongoDB にトリガーのような機能を実装できます。このコレクションをアクティブにするには、最初にレプリカセットをアクティブにする必要があります。このトピックの詳細は、https://docs.mongodb.com/manual/tutorial/deploy-replica-set/ を確認してください。

以下に、コンポーネントを使用して oplog コレクションを追跡する方法を示す Java DSL ベースのルートの例を示します。この特定のケースでは、データベース optlog_test 内のコレクション customers に影響を与えるイベントをフィルタリングしています。tailTrackIncreasingField はタイムスタンプフィールド ('ts') であることに注意してください。これは、TIMESTAMP 値で tailTrackingStrategy パラメーターを使用する必要があることを意味します。

import com.mongodb.BasicDBObject;
import com.mongodb.MongoClient;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mongodb.MongoDBTailTrackingEnum;
import org.apache.camel.main.Main;

import java.io.InputStream;

/**
 * For this to work you need to turn on the replica set
 * <p>
 * Commands to create a replica set:
 * <p>
 * rs.initiate( {
 * _id : "rs0",
 * members: [ { _id : 0, host : "localhost:27017" } ]
 * })
 */
public class MongoDbTracker {

    private final String database;

    private final String collection;

    private final String increasingField;

    private MongoDBTailTrackingEnum trackingStrategy;

    private int persistRecords = -1;

    private boolean persistenTailTracking;

    public MongoDbTracker(String database, String collection, String increasingField) {
        this.database = database;
        this.collection = collection;
        this.increasingField = increasingField;
    }

    public static void main(String[] args) throws Exception {
        final MongoDbTracker mongoDbTracker = new MongoDbTracker("local", "oplog.rs", "ts");
        mongoDbTracker.setTrackingStrategy(MongoDBTailTrackingEnum.TIMESTAMP);
        mongoDbTracker.setPersistRecords(5);
        mongoDbTracker.setPersistenTailTracking(true);
        mongoDbTracker.startRouter();
        // run until you terminate the JVM
        System.out.println("Starting Camel. Use ctrl + c to terminate the JVM.\n");

    }

    public void setTrackingStrategy(MongoDBTailTrackingEnum trackingStrategy) {
        this.trackingStrategy = trackingStrategy;
    }

    public void setPersistRecords(int persistRecords) {
        this.persistRecords = persistRecords;
    }

    public void setPersistenTailTracking(boolean persistenTailTracking) {
        this.persistenTailTracking = persistenTailTracking;
    }

    void startRouter() throws Exception {
        // create a Main instance
        Main main = new Main();
        main.bind(MongoConstants.CONN_NAME, new MongoClient("localhost", 27017));
        main.addRouteBuilder(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                getContext().getTypeConverterRegistry().addTypeConverter(InputStream.class, BasicDBObject.class,
                        new MongoToInputStreamConverter());
                from("mongodb://" + MongoConstants.CONN_NAME + "?database=" + database
                        + "&collection=" + collection
                        + "&persistentTailTracking=" + persistenTailTracking
                        + "&persistentId=trackerName" + "&tailTrackDb=local"
                        + "&tailTrackCollection=talendTailTracking"
                        + "&tailTrackField=lastTrackingValue"
                        + "&tailTrackIncreasingField=" + increasingField
                        + "&tailTrackingStrategy=" + trackingStrategy.toString()
                        + "&persistRecords=" + persistRecords
                        + "&cursorRegenerationDelay=1000")
                        .filter().jsonpath("$[?(@.ns=='optlog_test.customers')]")
                        .id("logger")
                        .to("log:logger?level=WARN")
                        .process(new Processor() {
                            public void process(Exchange exchange) throws Exception {
                                Message message = exchange.getIn();
                                System.out.println(message.getBody().toString());
                                exchange.getOut().setBody(message.getBody().toString());
                            }
                        });
            }
        });
        main.run();
    }
}