<commit_before>module.exports = exports = function SeamlessMongoosePlugin(schema){
  var buffer = Buffer();

  // data buffer abstraction - to move it under redis and make plugin stateless
  // and work in multithreaded envs
  function Buffer(){
    var databuf = {}, // buffer with stringified responses - per reqid
      queries = {}, // buffer with queries - per reqid
      clients = {}, // buffer with clients subscribed for a changes - per reqid
      requests = {}, // buffer with reqids - per document
      timestamp = {}; // buffer with timestamps of requests - per reqid

    var BUFFER_TTL = 21000;

    // buffer maintenance
    var _garbageCollector = setInterval(function() {
      var i, c = 0;
      for (i in timestamp) {
        if ((Date.now() - timestamp[i]) > BUFFER_TTL) {
          delete databuf[i];
          delete queries[i];
          delete clients[i];
          delete timestamp[i];
          // TODO: need to clear requests
        }
        else c++;
      }
      BUFFER_TTL = (-3 * c) + 21000;
    }, BUFFER_TTL);

    return {
      get(id,key){
        switch (key){
          case "data":
            return id?databuf[id]:databuf;
          case "query":
            return id?queries[id]:queries;
          case "clients":
            return id?clients[id]:clients;
          case "requests":
            return id?requests[id]:requests;
          case "timestamp":
            return id?timestamp[id]:timestamp;
          default:
            return (BUFFER_TTL-21000)/(-3); // count of unique requests
        }
      },
      set(id,key,v){
        switch (key){
          case "data":
            return v?databuf[id]=v:delete databuf[id];
          case "query":
            return v?queries[id]=v:delete queries[id];
          case "clients":
            return v?clients[id]=v:delete clients[id];
          case "requests":
            return v?requests[id]=v:delete requests[id];
          case "timestamp":
            return v?timestamp[id]=v:delete timestamp[id];
        }
      }
    }
  }

  function _resAdapter(data){
    data = (data instanceof Array)?data:[data];
    return data.filter(function(d){return !!(d._id || d._doc)})
      .map(function(d){ return d._doc || d });
  }

  function strfy(docs){
    return JSON.stringify((docs.length == 1)?docs[0]:docs);
  }

  function mapall(A,B){
    return (A||[]).reduce(function(C,a){
      return (B[a]||[]).reduce(function(c,b){
        return (c.indexOf(b)==-1)?(c.concat(b)):c;
      },C);
    },[]);
  }

  function RespondTo(responses,reqid){
    if (responses.send) responses = [responses];
    console.log(responses);
    return function(docs){
      var data;
      if (typeof docs === "string"){
        data = docs;
      }
      else {
        docs = _resAdapter(docs);
        data = buffer.set(reqid,"data",strfy(docs));
      }
      responses.forEach(function(res){
        if (res.type) res.type('application/json');
        res.send(data);
        if (!res.isWebsocket)
          SeamlessMongoosePlugin.deregisterClient(reqid,res);
      });
      if (docs instanceof Array) {
        docs.forEach(function(doc){
          var r = buffer.get(doc._id,"requests");
          if (r && r.indexOf && (r.indexOf(reqid) == -1)) {
            buffer.set(doc._id,"requests",r.push(reqid));
          }
        });
      }
    };
  }

  function HandleErrTo(response,reqid) {
    return function(err){
      SeamlessMongoosePlugin.deregisterClient(reqid,response);
      if (response.status) {
        response.status(500).send('Error querying: ' + err.toString());
      }
      else response.close(500, 'Error querying: ' + err.toString());
      console.error(err);
      return;
    };
  }

  SeamlessMongoosePlugin.registerClient = function(id, peer) {
    var c = buffer.get(id,"clients");
    if (c instanceof Array) {
      if (c.indexOf(peer) == -1) {
        c = c.concat(peer);
        buffer.set(id,"clients",c);
        return true;
      }
      return false;
    }
    else {
      c = [peer];
      console.log("add: "+buffer.set(id,"clients",c));
      return true;
    }
  };

  SeamlessMongoosePlugin.deregisterClient = function(id, peer) {
    var c = buffer.get(id,"clients");
    if (c && c.length) {
      var pos = c.indexOf(peer);
      if (pos >= 0) {
        c.splice(pos, 1);
        if (!c.length) {
          c=null;
        }
        console.log("remove: "+buffer.set(id,"clients",c));
        return true;
      }
    }
    return false;
  };

  schema.statics.notifyRegisteredClients = function(changed_docs_ids){
    var Model = this;
    console.log("changed",Date.now());
    return Promise.all(mapall(changed_docs_ids,buffer.get(undefined,"requests"))
    .map(function(reqid){ // get requests objects
      var q;
      console.log(reqid);
      if (q=buffer.get(reqid,"queries")){
        console.log(q);
        return Model.find(q)
          .then(RespondTo(buffer.get(reqid,"clients"),reqid));
      }
    }));
  }; // returns an array of promises

  schema.statics.getData = function(reqid,query){
    var b;
    if (b=buffer.get(reqid,"data")) return new Promise.resolve(b);<commit_msg>testing changes communication 2<commit_after>module.exports = exports = function SeamlessMongoosePlugin(schema){
  var buffer = Buffer();

  // data buffer abstraction - to move it under redis and make plugin stateless
  // and work in multithreaded envs
  function Buffer(){
    var databuf = {}, // buffer with stringified responses - per reqid
      queries = {}, // buffer with queries - per reqid
      clients = {}, // buffer with clients subscribed for a changes - per reqid
      requests = {}, // buffer with reqids - per document
      timestamp = {}; // buffer with timestamps of requests - per reqid

    var BUFFER_TTL = 21000;

    // buffer maintenance
    var _garbageCollector = setInterval(function() {
      var i, c = 0;
      for (i in timestamp) {
        if ((Date.now() - timestamp[i]) > BUFFER_TTL) {
          delete databuf[i];
          delete queries[i];
          delete clients[i];
          delete timestamp[i];
          // TODO: need to clear requests
        }
        else c++;
      }
      BUFFER_TTL = (-3 * c) + 21000;
    }, BUFFER_TTL);

    return {
      get(id,key){
        switch (key){
          case "data":
            return id?databuf[id]:databuf;
          case "query":
            return id?queries[id]:queries;
          case "clients":
            return id?clients[id]:clients;
          case "requests":
            return id?requests[id]:requests;
          case "timestamp":
            return id?timestamp[id]:timestamp;
          default:
            return (BUFFER_TTL-21000)/(-3); // count of unique requests
        }
      },
      set(id,key,v){
        switch (key){
          case "data":
            return v?databuf[id]=v:delete databuf[id];
          case "query":
            return v?queries[id]=v:delete queries[id];
          case "clients":
            return v?clients[id]=v:delete clients[id];
          case "requests":
            return v?requests[id]=v:delete requests[id];
          case "timestamp":
            return v?timestamp[id]=v:delete timestamp[id];
        }
      }
    }
  }

  function _resAdapter(data){
    data = (data instanceof Array)?data:[data];
    return data.filter(function(d){return !!(d._id || d._doc)})
      .map(function(d){ return d._doc || d });
  }

  function strfy(docs){
    return JSON.stringify((docs.length == 1)?docs[0]:docs);
  }

  function mapall(A,B){
    return (A||[]).reduce(function(C,a){
      return (B[a]||[]).reduce(function(c,b){
        return (c.indexOf(b)==-1)?(c.concat(b)):c;
      },C);
    },[]);
  }

  function RespondTo(responses,reqid){
    if (responses.send) responses = [responses];
    console.log(responses);
    return function(docs){
      var data;
      if (typeof docs === "string"){
        data = docs;
      }
      else {
        docs = _resAdapter(docs);
        data = buffer.set(reqid,"data",strfy(docs));
      }
      responses.forEach(function(res){
        if (res.type) res.type('application/json');
        res.send(data);
        if (!res.isWebsocket)
          SeamlessMongoosePlugin.deregisterClient(reqid,res);
      });
      if (docs instanceof Array) {
        docs.forEach(function(doc){
          var r = buffer.get(doc._id,"requests");
          if (r && r.indexOf && (r.indexOf(reqid) == -1)) {
            buffer.set(doc._id,"requests",r.push(reqid));
          }
        });
      }
    };
  }

  function HandleErrTo(response,reqid) {
    return function(err){
      SeamlessMongoosePlugin.deregisterClient(reqid,response);
      if (response.status) {
        response.status(500).send('Error querying: ' + err.toString());
      }
      else response.close(500, 'Error querying: ' + err.toString());
      console.error(err);
      return;
    };
  }

  SeamlessMongoosePlugin.registerClient = function(id, peer) {
    var c = buffer.get(id,"clients");
    if (c instanceof Array) {
      if (c.indexOf(peer) == -1) {
        c = c.concat(peer);
        buffer.set(id,"clients",c);
        return true;
      }
      return false;
    }
    else {
      c = [peer];
      console.log("add: "+buffer.set(id,"clients",c));
      return true;
    }
  };

  SeamlessMongoosePlugin.deregisterClient = function(id, peer) {
    var c = buffer.get(id,"clients");
    if (c && c.length) {
      var pos = c.indexOf(peer);
      if (pos >= 0) {
        c.splice(pos, 1);
        if (!c.length) {
          c=null;
        }
        console.log("remove: "+buffer.set(id,"clients",c));
        return true;
      }
    }
    return false;
  };

  schema.statics.notifyRegisteredClients = function(changed_docs_ids){
    var Model = this;
    console.log("changed",Date.now());
    return Promise.all(
      mapall(changed_docs_ids,buffer.get(undefined,"requests"))
      .map(function(reqid){ // get requests objects
        var q;
        console.log(reqid);
        if (q=buffer.get(reqid,"queries")){
          console.log(q);
          return Model.find(q)
            .then(RespondTo(buffer.get(reqid,"clients"),reqid));
        }
      })
    );
  }; // returns an array of promises

  schema.statics.getData = function(reqid,query){
    var b;
    if (b=buffer.get(reqid,"data")) return new Promise.resolve(b);
{'fraction_non_alphanumeric': 0.14877152253820855, 'fraction_numerical': 0.006094022054556007, 'mean_word_length': 2.1949938195302843, 'pattern_counts': {'":': 20, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2856066', 'n_tokens_mistral': 3304, 'n_tokens_neox': 3211, 'n_words': 683}