event-buffer-factory.js
function sendEvents() {

    logger.logEnter('sendEvents with eventCount ' + eventCount);

    if (eventCount > 0) {

      // If there are fewer than 10 events, we wait for more to arrive
      // before sending them.  Only delay sending them once so they
      // don't get too old.
      if (eventCount < 10 && canDelaySend) {

        // Wait for more events.
        canDelaySend = false;

        logger.logExit('sendEvents, wait for more events');
        return;
      }

      // Post the events.
      canDelaySend = true;

      // Ensure they're valid.
      validate();

      if (eventCount === 0) {

        // All events were invalid.  We're done.
        logger.logExit('sendEvents, all events were invalid');
        return;
      }

      try {
        // Use one of the three APIs: there's one for a single
        // event, one for multiple events of the same type, and
        // one for a mix of types.
        var path,
          request,
          eventData,
          failureCallback,
          retryEventCount, retryEventData, retryBuffer;

        if (buffer.length === 1) {
          // They're all the same type.
          eventData = buffer[0];

          if(runningInPublicEnv) {
            path = '/imfmobileanalytics/proxy/v1/apps/' + settings.AZFilterAppId + '/' + eventData[0];
          } else {
            path = '/' + eventData[0];
          }

          if (eventCount > 1) {
            path += '/_bulk';
          }
                    
          if(runningInPublicEnv) {
            // Deal with expiration of access token by refreshing token and retry the request.
            retryEventCount = eventCount;
            retryEventData = eventData;
            
            failureCallback = function(failureResponse, failureMessage) {
              emitter.emit('error', failureMessage);
              
              if(failureResponse.statusCode === 401) {
                logger.log('Received 401 Unathorized response while posting event(s) of same type.  Refresh access token and retry the post.');
                
                environment.refreshProxyAccessToken();
                var retryRequest = createLogRequest(logger, utils, path, environment, emitter, null);
                
                if (retryEventCount > 1) {
                  for (var i = 1; i < retryEventData.length; i++) {
                    retryRequest.write('{"create":{}}\n');
                    retryRequest.write(JSON.stringify(retryEventData[i]) + '\n');
                  }
                  retryRequest.end();
                } else {
                  retryRequest.end(JSON.stringify(retryEventData[1]));
                }
              }
            };
          }
          
          request = createLogRequest(logger, utils, path, environment, emitter, failureCallback);

          // If this request processing logic is changed, then retryRequest processing above should also be changed.
          if (eventCount > 1) {
            for (var i = 1; i < eventData.length; i++) {
              request.write('{"create":{}}\n');
              request.write(JSON.stringify(eventData[i]) + '\n');
            }
            request.end();
          } else {
            request.end(JSON.stringify(eventData[1]));
          }

        } else {

          // There's more than one type of event.
          if(runningInPublicEnv) {
            path = '/imfmobileanalytics/proxy/v1/apps/' + settings.AZFilterAppId + '/_bulk';
          } else {
            path = '/_bulk';
          }
          
          if(runningInPublicEnv) {
            // Deal with expiration of access token by refreshing token and retry the request.
            retryBuffer = buffer.slice(0);
            
            failureCallback = function(failureResponse, failureMessage) {
              emitter.emit('error', failureMessage);
              
              if(failureResponse.statusCode === 401) {
                logger.log('Received 401 Unathorized response while posting event(s) of different types.  Refresh access token and retry the post.');
                
                environment.refreshProxyAccessToken();
                var retryRequest = createLogRequest(logger, utils, path, environment, emitter, null);
                
                for (var x = 0; x < retryBuffer.length; x++) {
                  retryEventData = retryBuffer[x];
                  var createObject = '{"create":{"_type":"' + retryEventData[0] + '"}}\n';
                  for (var j = 1; j < retryEventData.length; j++) {
                    retryRequest.write(createObject);
                    retryRequest.write(JSON.stringify(retryEventData[j]) + '\n');
                  }
                }

                retryRequest.end();
              }
            };
          }
          
          request = createLogRequest(logger, utils, path, environment, emitter, failureCallback);

          // If this request processing logic is changed, then retryRequest processing above should also be changed.
          for (var x = 0; x < buffer.length; x++) {
            eventData = buffer[x];
            var createObject = '{"create":{"_type":"' + eventData[0] + '"}}\n';
            for (var j = 1; j < eventData.length; j++) {
              request.write(createObject);
              request.write(JSON.stringify(eventData[j]) + '\n');
            }
          }

          request.end();
        }
      } catch (thrown) {
        emitter.emit('error', 'While reporting events, caught ' + thrown);
      }

      buffer.clear();
    }

    logger.logExit('sendEvents');
  }
function sendEvents() {

    logger.logEnter('sendEvents with eventCount ' + eventCount);

    if (eventCount > 0) {

      // If there are fewer than 10 events, we wait for more to arrive
      // before sending them.  Only delay sending them once so they
      // don't get too old.
      if (eventCount < 10 && canDelaySend) {

        // Wait for more events.
        canDelaySend = false;

        logger.logExit('sendEvents, wait for more events');
        return;
      }

      // Post the events.
      canDelaySend = true;

      // Ensure they're valid.
      validate();

      if (eventCount === 0) {

        // All events were invalid.  We're done.
        logger.logExit('sendEvents, all events were invalid');
        return;
      }

      try {
        // Use one of the three APIs: there's one for a single
        // event, one for multiple events of the same type, and
        // one for a mix of types.
        var path,
          request,
          eventData,
          failureCallback,
          retryEventCount, retryEventData, retryBuffer;

        if (buffer.length === 1) {
          // They're all the same type.
          eventData = buffer[0];

          if(runningInPublicEnv) {
            path = '/imfmobileanalytics/proxy/v1/apps/' + settings.AZFilterAppId + '/' + eventData[0];
          } else {
            path = '/' + eventData[0];
          }

          if (eventCount > 1) {
            path += '/_bulk';
          }
                    
          if(runningInPublicEnv) {
            // Deal with expiration of access token by refreshing token and retry the request.
            retryEventCount = eventCount;
            retryEventData = eventData;
            
            failureCallback = function(failureResponse, failureMessage) {
              emitter.emit('error', failureMessage);
              
              if(failureResponse.statusCode === 401) {
                logger.log('Received 401 Unathorized response while posting event(s) of same type.  Refresh access token and retry the post.');
                
                environment.refreshProxyAccessToken();
                var retryRequest = createLogRequest(logger, utils, path, environment, emitter, null);
                
                if (retryEventCount > 1) {
                  for (var i = 1; i < retryEventData.length; i++) {
                    retryRequest.write('{"create":{}}\n');
                    retryRequest.write(JSON.stringify(retryEventData[i]) + '\n');
                  }
                  retryRequest.end();
                } else {
                  retryRequest.end(JSON.stringify(retryEventData[1]));
                }
              }
            };
          }
          
          request = createLogRequest(logger, utils, path, environment, emitter, failureCallback);

          // If this request processing logic is changed, then retryRequest processing above should also be changed.
          if (eventCount > 1) {
            for (var i = 1; i < eventData.length; i++) {
              request.write('{"create":{}}\n');
              request.write(JSON.stringify(eventData[i]) + '\n');
            }
            request.end();
          } else {
            request.end(JSON.stringify(eventData[1]));
          }

        } else {

          // There's more than one type of event.
          if(runningInPublicEnv) {
            path = '/imfmobileanalytics/proxy/v1/apps/' + settings.AZFilterAppId + '/_bulk';
          } else {
            path = '/_bulk';
          }
          
          if(runningInPublicEnv) {
            // Deal with expiration of access token by refreshing token and retry the request.
            retryBuffer = buffer.slice(0);
            
            failureCallback = function(failureResponse, failureMessage) {
              emitter.emit('error', failureMessage);
              
              if(failureResponse.statusCode === 401) {
                logger.log('Received 401 Unathorized response while posting event(s) of different types.  Refresh access token and retry the post.');
                
                environment.refreshProxyAccessToken();
                var retryRequest = createLogRequest(logger, utils, path, environment, emitter, null);
                
                for (var x = 0; x < retryBuffer.length; x++) {
                  retryEventData = retryBuffer[x];
                  var createObject = '{"create":{"_type":"' + retryEventData[0] + '"}}\n';
                  for (var j = 1; j < retryEventData.length; j++) {
                    retryRequest.write(createObject);
                    retryRequest.write(JSON.stringify(retryEventData[j]) + '\n');
                  }
                }

                retryRequest.end();
              }
            };
          }
          
          request = createLogRequest(logger, utils, path, environment, emitter, failureCallback);

          // If this request processing logic is changed, then retryRequest processing above should also be changed.
          for (var x = 0; x < buffer.length; x++) {
            eventData = buffer[x];
            var createObject = '{"create":{"_type":"' + eventData[0] + '"}}\n';
            for (var j = 1; j < eventData.length; j++) {
              request.write(createObject);
              request.write(JSON.stringify(eventData[j]) + '\n');
            }
          }

          request.end();
        }
      } catch (thrown) {
        emitter.emit('error', 'While reporting events, caught ' + thrown);
      }

      buffer.clear();
    }

    logger.logExit('sendEvents');
  }
function sendEvents ( ) { logger . logEnter ( 'sendEvents with eventCount ' + eventCount ) ; if ( eventCount > 0 ) { // If there are fewer than 10 events, we wait for more to arrive // before sending them.  Only delay sending them once so they // don't get too old. if ( eventCount < 10 && canDelaySend ) { // Wait for more events. canDelaySend = false ; logger . logExit ( 'sendEvents, wait for more events' ) ; return ; } // Post the events. canDelaySend = true ; // Ensure they're valid. validate ( ) ; if ( eventCount === 0 ) { // All events were invalid.  We're done. logger . logExit ( 'sendEvents, all events were invalid' ) ; return ; } try { // Use one of the three APIs: there's one for a single // event, one for multiple events of the same type, and // one for a mix of types. var path , request , eventData , failureCallback , retryEventCount , retryEventData , retryBuffer ; if ( buffer . length === 1 ) { // They're all the same type. eventData = buffer [ 0 ] ; if ( runningInPublicEnv ) { path = '/imfmobileanalytics/proxy/v1/apps/' + settings . AZFilterAppId + '/' + eventData [ 0 ] ; } else { path = '/' + eventData [ 0 ] ; } if ( eventCount > 1 ) { path += '/_bulk' ; } if ( runningInPublicEnv ) { // Deal with expiration of access token by refreshing token and retry the request. retryEventCount = eventCount ; retryEventData = eventData ; failureCallback = function ( failureResponse , failureMessage ) { emitter . emit ( 'error' , failureMessage ) ; if ( failureResponse . statusCode === 401 ) { logger . log ( 'Received 401 Unathorized response while posting event(s) of same type.  Refresh access token and retry the post.' ) ; environment . refreshProxyAccessToken ( ) ; var retryRequest = createLogRequest ( logger , utils , path , environment , emitter , null ) ; if ( retryEventCount > 1 ) { for ( var i = 1 ; i < retryEventData . length ; i ++ ) { retryRequest . write ( '{"create":{}}\n' ) ; retryRequest . write ( JSON . stringify ( retryEventData [ i ] ) + '\n' ) ; } retryRequest . end ( ) ; } else { retryRequest . end ( JSON . stringify ( retryEventData [ 1 ] ) ) ; } } } ; } request = createLogRequest ( logger , utils , path , environment , emitter , failureCallback ) ; // If this request processing logic is changed, then retryRequest processing above should also be changed. if ( eventCount > 1 ) { for ( var i = 1 ; i < eventData . length ; i ++ ) { request . write ( '{"create":{}}\n' ) ; request . write ( JSON . stringify ( eventData [ i ] ) + '\n' ) ; } request . end ( ) ; } else { request . end ( JSON . stringify ( eventData [ 1 ] ) ) ; } } else { // There's more than one type of event. if ( runningInPublicEnv ) { path = '/imfmobileanalytics/proxy/v1/apps/' + settings . AZFilterAppId + '/_bulk' ; } else { path = '/_bulk' ; } if ( runningInPublicEnv ) { // Deal with expiration of access token by refreshing token and retry the request. retryBuffer = buffer . slice ( 0 ) ; failureCallback = function ( failureResponse , failureMessage ) { emitter . emit ( 'error' , failureMessage ) ; if ( failureResponse . statusCode === 401 ) { logger . log ( 'Received 401 Unathorized response while posting event(s) of different types.  Refresh access token and retry the post.' ) ; environment . refreshProxyAccessToken ( ) ; var retryRequest = createLogRequest ( logger , utils , path , environment , emitter , null ) ; for ( var x = 0 ; x < retryBuffer . length ; x ++ ) { retryEventData = retryBuffer [ x ] ; var createObject = '{"create":{"_type":"' + retryEventData [ 0 ] + '"}}\n' ; for ( var j = 1 ; j < retryEventData . length ; j ++ ) { retryRequest . write ( createObject ) ; retryRequest . write ( JSON . stringify ( retryEventData [ j ] ) + '\n' ) ; } } retryRequest . end ( ) ; } } ; } request = createLogRequest ( logger , utils , path , environment , emitter , failureCallback ) ; // If this request processing logic is changed, then retryRequest processing above should also be changed. for ( var x = 0 ; x < buffer . length ; x ++ ) { eventData = buffer [ x ] ; var createObject = '{"create":{"_type":"' + eventData [ 0 ] + '"}}\n' ; for ( var j = 1 ; j < eventData . length ; j ++ ) { request . write ( createObject ) ; request . write ( JSON . stringify ( eventData [ j ] ) + '\n' ) ; } } request . end ( ) ; } } catch ( thrown ) { emitter . emit ( 'error' , 'While reporting events, caught ' + thrown ) ; } buffer . clear ( ) ; } logger . logExit ( 'sendEvents' ) ; }
This function will run every so often to send the buffered events to Elasticsearch.
This function will run every so often to send the buffered events to Elasticsearch .