martin-cowie

package com.acme;

import java.lang.management.ManagementFactory;

import javax.management.MBeanServer;

import com.pushtechnology.diffusion.api.APIException;
import com.pushtechnology.diffusion.api.Logs;
import com.pushtechnology.diffusion.api.publisher.ServerShutdownHook;
import com.pushtechnology.diffusion.api.publisher.ServerStartupHook;

martin-cowie

final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();

Diskspace mbean = new Diskspace(root);
try {
    ObjectName objectName = new ObjectName( "com.acme.diskspace:name=foo" );
    mbs.registerMBean( mbean, objectName );
} catch ( Exception ex ) {
    throw new Error( String.format( "Cannot registry MBean at "%s"", mbean.getObjectName() ), ex );
}

martin-cowie

#!/usr/bin/env node
var request = require('request');
var MongoClient = require('mongodb').MongoClient;
var assert = require('assert');

const truefxURL = "http://webrates.truefx.com/rates/connect.html";
const pollFreq = 500;

const mongoURL = (process.argv.length == 3) ? process.argv[2] : 'mongodb://localhost/truefx';
const mongoCollectionName = "currencies";

martin-cowie

% mongo localhost:20001

MongoDB shell version: 3.2.6

connecting to: localhost:20000/test
testReplSet:PRIMARY> use someDB

switched to db someDB
testReplSet:PRIMARY> for (i=0; i<100; i++) { db.someCollection.insert({count: i}) }

WriteResult({ "nInserted" : 1 })

martin-cowie

% mongo --nodb
MongoDB shell version: 3.2.6
> replicaSet = new ReplSetTest({"nodes" : 1, "nodeOptions": {"dbpath": "/tmp/mongodb"} });
> replicaSet.startSet()
> replicaSet.initiate();

martin-cowie

/**
 * Subscribe to {@code local.oplog.rs}, listening for relevant changes.
 */
private void relayChanges(long timeNow) {
    LOG.info("Relaying changes {} to {}", oplog.getNamespace(), topicRoot);

    final BsonTimestamp now = new BsonTimestamp((int) (timeNow / 1000), 1);
    final Bson filter = and(
        gt("ts", now),
        or(

martin-cowie

/**
 * Enumerate all documents in the collection, then relay changes.
 */
private void run() {
    final long timeNow = System.currentTimeMillis();
    long topicCount = 0;
    LOG.info("Transcribing topics from {} to {}", collection.getNamespace(),
        topicRoot);
    try (
        final MongoCursor<Document> cursor = collection.find().iterator()) {

martin-cowie

/**
 * Build an Adapter.
 * <P>
 *
 * @param session Configured Diffusion session.
 * @param collection Source of documents for reflection.
 * @param oplog Collection holding the MongoDB replication operation log.
 * @param topicRoot Topic under which other topics are to be created.
 * @return a ready-to-use Adapter
 * @throws InterruptedException if the current thread was interrupted while waiting for update rights

martin-cowie

/**
 * Command line bootstrap method.
 */
public static void main(String[] argv) throws Exception {

    final CommandlineArgs args = new CommandlineArgs();
    final JCommander jc = new JCommander(args);
    try {
        jc.parse(argv);
    }