Skip to content

Instantly share code, notes, and snippets.

@laser

laser/streams.js

Last active March 27, 2018 00:30
Show Gist options
  • Save laser/91e8fe32b2682d3ff39f to your computer and use it in GitHub Desktop.
Save laser/91e8fe32b2682d3ff39f to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 4 additions and 0 deletions.
    4 changes: 4 additions & 0 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -7,6 +7,10 @@ var request = require('request-json'),
    client = request.newClient('http://localhost:3000/api/'),
    db = require('mongodb').MongoClient;

    ///////////////////////////////////////////////////////////
    // this is all boilerplate; scroll down for demo code
    //

    app.get('/', function(req, res) {
    res.sendFile('./static/index.html');
    });
  2. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 15 additions and 2 deletions.
    17 changes: 15 additions & 2 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -4,7 +4,8 @@ var request = require('request-json'),
    http = require('http').Server(app),
    io = require('socket.io'),
    Bacon = require('baconjs').Bacon,
    client = request.newClient('http://localhost:3000/api/');
    client = request.newClient('http://localhost:3000/api/'),
    db = require('mongodb').MongoClient;

    app.get('/', function(req, res) {
    res.sendFile('./static/index.html');
    @@ -34,9 +35,16 @@ http.listen(3000, function() {

    io = io(http);

    function getFunFact(callback) {
    client.connect('mongodb://127.0.0.1:27017/library', function(err, db) {
    db.collection('facts').findOne(callback);
    });
    }

    ///////////////////////////////////////////////////////////
    // demo impl below
    //

    var connections = Bacon.fromBinder(function(sink) {
    io.on('connection', sink)
    });
    @@ -58,7 +66,12 @@ var funFact = currentMessages
    .filter(function(acc) {
    return acc.length === 20;
    })
    .map("Did you know...?");
    .flatMap(function() {
    return Bacon.fromNodeCallback(getFunFact);
    })
    .map(function(fact) {
    return "Did you know that: " + fact.text;
    });

    var entries = currentMessages
    .filter(function(acc) {
  3. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 19 additions and 14 deletions.
    33 changes: 19 additions & 14 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -52,23 +52,28 @@ var messages = connections.flatMap(function(socket) {
    var currentMessages = messages
    .scan([], function(acc, message) {
    return acc.length === 20 ? [] : acc.concat(message);
    })
    .filter(function(acc) {
    return acc.length === 20;
    });

    var funFact = currentMessages.map("Did you know...?");
    var funFact = currentMessages
    .filter(function(acc) {
    return acc.length === 20;
    })
    .map("Did you know...?");

    var entries = currentMessages.flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    var entries = currentMessages
    .filter(function(acc) {
    return acc.length === 20;
    })
    .flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    });

    connections.onValue(function(socket) {
  4. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 35 additions and 1 deletion.
    36 changes: 35 additions & 1 deletion streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -1,8 +1,42 @@
    var request = require('request-json'),
    io = require('./server'),
    express = require('express'),
    app = express(),
    http = require('http').Server(app),
    io = require('socket.io'),
    Bacon = require('baconjs').Bacon,
    client = request.newClient('http://localhost:3000/api/');

    app.get('/', function(req, res) {
    res.sendFile('./static/index.html');
    });

    app.post('/api/log', function(req, res) {
    var x = ((Math.random() * 10) > 5) ? "NOT\nJSON" : true;
    res.send(x);
    });

    app.get('/api/weather', function(req, res) {
    res.send({
    "main": {
    "temp": (Math.round(Math.random() * 1000) / 10)
    },
    "weather": [{
    "main": "Cloudy"
    }]
    });
    });

    app.use(express.static('./static'));

    http.listen(3000, function() {
    console.log(new Date(), 'started server');
    });

    io = io(http);

    ///////////////////////////////////////////////////////////
    // demo impl below
    //
    var connections = Bacon.fromBinder(function(sink) {
    io.on('connection', sink)
    });
  5. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 2 additions and 2 deletions.
    4 changes: 2 additions & 2 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -17,10 +17,10 @@ var messages = connections.flatMap(function(socket) {

    var currentMessages = messages
    .scan([], function(acc, message) {
    return acc.length === 5 ? [] : acc.concat(message);
    return acc.length === 20 ? [] : acc.concat(message);
    })
    .filter(function(acc) {
    return acc.length === 5;
    return acc.length === 20;
    });

    var funFact = currentMessages.map("Did you know...?");
  6. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 16 additions and 21 deletions.
    37 changes: 16 additions & 21 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -17,29 +17,24 @@ var messages = connections.flatMap(function(socket) {

    var currentMessages = messages
    .scan([], function(acc, message) {
    return acc.length === 20 ? [] : acc.concat(message);
    });

    var funFact = currentMessages
    .filter(function(acc) {
    return acc.length === 20;
    return acc.length === 5 ? [] : acc.concat(message);
    })
    .map("Did you know...?");

    var entries = currentMessages
    .filter(function(acc) {
    return acc.length === 20;
    })
    .flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    return acc.length === 5;
    });

    var funFact = currentMessages.map("Did you know...?");

    var entries = currentMessages.flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    });

    connections.onValue(function(socket) {
  7. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 3 additions and 3 deletions.
    6 changes: 3 additions & 3 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -17,18 +17,18 @@ var messages = connections.flatMap(function(socket) {

    var currentMessages = messages
    .scan([], function(acc, message) {
    return acc.length === 5 ? [] : acc.concat(message);
    return acc.length === 20 ? [] : acc.concat(message);
    });

    var funFact = currentMessages
    .filter(function(acc) {
    return acc.length === 5;
    return acc.length === 20;
    })
    .map("Did you know...?");

    var entries = currentMessages
    .filter(function(acc) {
    return acc.length === 5;
    return acc.length === 20;
    })
    .flatMap(function(messages) {
    return Bacon.retry({
  8. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 19 additions and 14 deletions.
    33 changes: 19 additions & 14 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -18,23 +18,28 @@ var messages = connections.flatMap(function(socket) {
    var currentMessages = messages
    .scan([], function(acc, message) {
    return acc.length === 5 ? [] : acc.concat(message);
    })
    });

    var funFact = currentMessages
    .filter(function(acc) {
    return acc.length === 5;
    });
    })
    .map("Did you know...?");

    var funFact = currentMessages.map("Did you know...?");

    var entries = currentMessages.flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    var entries = currentMessages
    .filter(function(acc) {
    return acc.length === 5;
    })
    .flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    });

    connections.onValue(function(socket) {
  9. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 11 additions and 12 deletions.
    23 changes: 11 additions & 12 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -25,17 +25,16 @@ var currentMessages = messages

    var funFact = currentMessages.map("Did you know...?");

    var entries = currentMessages
    .flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    var entries = currentMessages.flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    });

    connections.onValue(function(socket) {
    @@ -52,4 +51,4 @@ funFact.onValue(function(fact) {

    entries.onValue(function() {
    // deliberate no-op
    });
    });
  10. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 8 additions and 10 deletions.
    18 changes: 8 additions & 10 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -15,13 +15,15 @@ var messages = connections.flatMap(function(socket) {
    });
    });

    var currentMessages = messages.scan([], function(acc, message) {
    return acc.length === 20 ? [] : acc.concat(message);
    });
    var currentMessages = messages
    .scan([], function(acc, message) {
    return acc.length === 5 ? [] : acc.concat(message);
    })
    .filter(function(acc) {
    return acc.length === 5;
    });

    var funFact = currentMessages
    .filter(function(acc) { return acc.length === 20 })
    .map("Did you know...?");
    var funFact = currentMessages.map("Did you know...?");

    var entries = currentMessages
    .flatMap(function(messages) {
    @@ -50,8 +52,4 @@ funFact.onValue(function(fact) {

    entries.onValue(function() {
    // deliberate no-op
    });

    entries.onError(function(err) {
    throw err;
    });
  11. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 6 additions and 2 deletions.
    8 changes: 6 additions & 2 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -16,11 +16,11 @@ var messages = connections.flatMap(function(socket) {
    });

    var currentMessages = messages.scan([], function(acc, message) {
    return acc.length === 520? [] : acc.concat(message);
    return acc.length === 20 ? [] : acc.concat(message);
    });

    var funFact = currentMessages
    .filter(function(acc) { return acc.length && acc.length % 5 === 0 })
    .filter(function(acc) { return acc.length === 20 })
    .map("Did you know...?");

    var entries = currentMessages
    @@ -50,4 +50,8 @@ funFact.onValue(function(fact) {

    entries.onValue(function() {
    // deliberate no-op
    });

    entries.onError(function(err) {
    throw err;
    });
  12. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -16,7 +16,7 @@ var messages = connections.flatMap(function(socket) {
    });

    var currentMessages = messages.scan([], function(acc, message) {
    return acc.length === 5 ? [] : acc.concat(message);
    return acc.length === 520? [] : acc.concat(message);
    });

    var funFact = currentMessages
  13. Erin Swenson-Healey revised this gist Sep 24, 2014. 1 changed file with 0 additions and 28 deletions.
    28 changes: 0 additions & 28 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -7,30 +7,6 @@ var connections = Bacon.fromBinder(function(sink) {
    io.on('connection', sink)
    });

    //var currentWeather = Bacon
    //.fromBinder(function(sink) {
    //function poll() {
    //client.get('weather', function(err, data) {
    //sink(err ? new Bacon.Error(err) : new Bacon.Next(data));
    //setTimeout(poll, 2000);
    //});
    //}

    //poll();
    //})
    //.map('.body')
    //.map(JSON.parse)
    //.map(function(body) {
    //return body.weather[0].main + ', ' + body.main.temp + 'F';
    //})
    //.toProperty();

    //var greetings = currentWeather
    //.sampledBy(connections, function(weather, socket) {
    //var s = 'Welcome! Current weather is: ' + weather;
    //return { txt: s, recip: socket }
    //});

    var messages = connections.flatMap(function(socket) {
    return Bacon.fromBinder(function(sink) {
    socket.on('message', function(txt) {
    @@ -64,10 +40,6 @@ connections.onValue(function(socket) {
    socket.broadcast.emit('message', 'CONN: ' + socket.id);
    });

    //greetings.onValue(function(message) {
    //message.recip.send(message.txt);
    //});

    messages.onValue(function(message) {
    io.emit('message', '' + message.author.id + ': ' + message.txt);
    });
  14. Erin Swenson-Healey created this gist Sep 24, 2014.
    81 changes: 81 additions & 0 deletions streams.js
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,81 @@
    var request = require('request-json'),
    io = require('./server'),
    Bacon = require('baconjs').Bacon,
    client = request.newClient('http://localhost:3000/api/');

    var connections = Bacon.fromBinder(function(sink) {
    io.on('connection', sink)
    });

    //var currentWeather = Bacon
    //.fromBinder(function(sink) {
    //function poll() {
    //client.get('weather', function(err, data) {
    //sink(err ? new Bacon.Error(err) : new Bacon.Next(data));
    //setTimeout(poll, 2000);
    //});
    //}

    //poll();
    //})
    //.map('.body')
    //.map(JSON.parse)
    //.map(function(body) {
    //return body.weather[0].main + ', ' + body.main.temp + 'F';
    //})
    //.toProperty();

    //var greetings = currentWeather
    //.sampledBy(connections, function(weather, socket) {
    //var s = 'Welcome! Current weather is: ' + weather;
    //return { txt: s, recip: socket }
    //});

    var messages = connections.flatMap(function(socket) {
    return Bacon.fromBinder(function(sink) {
    socket.on('message', function(txt) {
    sink({ author: socket, txt: txt });
    });
    });
    });

    var currentMessages = messages.scan([], function(acc, message) {
    return acc.length === 5 ? [] : acc.concat(message);
    });

    var funFact = currentMessages
    .filter(function(acc) { return acc.length && acc.length % 5 === 0 })
    .map("Did you know...?");

    var entries = currentMessages
    .flatMap(function(messages) {
    return Bacon.retry({
    retries: 10,
    delay: function() { return 100; },
    source: function() {
    return Bacon.fromNodeCallback(client, 'post', 'log', {
    messages: messages
    });
    }
    });
    });

    connections.onValue(function(socket) {
    socket.broadcast.emit('message', 'CONN: ' + socket.id);
    });

    //greetings.onValue(function(message) {
    //message.recip.send(message.txt);
    //});

    messages.onValue(function(message) {
    io.emit('message', '' + message.author.id + ': ' + message.txt);
    });

    funFact.onValue(function(fact) {
    io.emit('message', fact);
    });

    entries.onValue(function() {
    // deliberate no-op
    });