GELFManager - Handles GELF messages

Example:

var manager = new GELFManager();
manager.on('message', function(msg) { console.log(msg); });
manager.feed(rawUDPMessage);
961969
function GELFManager(options) {

  if (typeof options === 'undefined') options = {};
  for (k in GELFManager.options)
    if (typeof options[k] === 'undefined')
      options[k] = GELFManager.options[k];

  this.debug        = options.debug;
  this.chunkTimeout = options.chunkTimeout;
  this.gcTimeout    = options.gcTimeout;

  EventEmitter.call(this);
  this.chunksPool = {};

  process.nextTick(this._gc.bind(this));

}
272520
function parseMessage(msg, jsonEncode) {
    if (typeof msg === 'string') {
        msg = { data: msg };
    }

    var output = '';
    if (msg.event) {
        output += 'event: ' + msg.event + '\n';
    }

    if (msg.retry) {
        output += 'retry: ' + msg.retry + '\n';
    }

    if (msg.id) {
        output += 'id: ' + msg.id + '\n';
    }

    var data = msg.data || '';
    if (jsonEncode) {
        data = JSON.stringify(data);
    }

    output += parseTextData(data);

    return output;
}
868706
public GelfMessage createGelfMessage(LogEvent logEvent) {

        GelfMessageBuilder builder = builders != null ? builders.get().recycle() : newInstance();

        Throwable throwable = logEvent.getThrowable();
        String message = logEvent.getMessage();

        if (GelfMessage.isEmpty(message) && throwable != null) {
            message = throwable.toString();
        }

        String shortMessage = message;
        if (message.length() > MAX_SHORT_MESSAGE_LENGTH) {
            shortMessage = message.substring(0, MAX_SHORT_MESSAGE_LENGTH - 1);
        }

        builder.withShortMessage(shortMessage).withFullMessage(message).withJavaTimestamp(logEvent.getLogTimestamp());
        builder.withLevel(logEvent.getSyslogLevel());
        builder.withVersion(getVersion());
        builder.withAdditionalFieldTypes(additionalFieldTypes);

        for (MessageField field : fields) {

            if (!isIncludeLocation() && field instanceof LogMessageField) {

                LogMessageField messageField = (LogMessageField) field;
                if (SOURCE_FIELDS.contains(messageField.getNamedLogField())) {
                    continue;
                }
            }

            Values values = getValues(logEvent, field);
            if (values == null || !values.hasValues()) {
                continue;
            }

            for (String entryName : values.getEntryNames()) {
                String value = values.getValue(entryName);
                if (value == null) {
                    continue;

                }
                builder.withField(entryName, value);
            }
        }

        if (stackTraceExtraction.isEnabled() && throwable != null) {
            addStackTrace(throwable, builder);
        }

        if (includeLogMessageParameters && logEvent.getParameters() != null) {
            for (int i = 0; i < logEvent.getParameters().length; i++) {
                Object param = logEvent.getParameters()[i];
                builder.withField(FIELD_MESSAGE_PARAM + i, "" + param);
            }
        }

        builder.withHost(getOriginHost());

        if (null != facility) {
            builder.withFacility(facility);
        }

        builder.withMaximumMessageSize(maximumMessageSize);
        return builder.build();
    }
980733
function(messageName, handler) {
		var agora = (this.env ||  this).data.agora,
			self = this;
		var suphandler = function() {
			handler.apply(self, arguments);
		};
		agora.on(messageName, suphandler);
		(this.binds = this.binds ||  []).push(function() {
			agora.off(messageName, suphandler);
		});
		return this;
	}
1310757
function logMessage (message) {
  if (!process.send) {
    console.log(message)
  } else {
    process.send(message)
  }
}
118755
function(_target,_message){
                var _from = _message.from;
                while(!!_target){
                    if (_target._$getPath()!=_from){
                        _doSendMessage(_target,_message);
                    }
                    _target = _target._$getParent();
                }
            }
848670
function (msg) {
            var msgHandlers;
            if (!msg.method) {
                // no message type, ignoring it
                // TODO: should we trigger a generic event?
                console.log("[Brackets LiveDev] Received message without method.");
                return;
            }
            // get handlers for msg.method
            msgHandlers = this.handlers[msg.method];

            if (msgHandlers && msgHandlers.length > 0) {
                // invoke handlers with the received message
                msgHandlers.forEach(function (handler) {
                    try {
                        // TODO: check which context should be used to call handlers here.
                        handler(msg);
                        return;
                    } catch (e) {
                        console.log("[Brackets LiveDev] Error executing a handler for " + msg.method);
                        console.log(e.stack);
                        return;
                    }
                });
            } else {
                // no subscribers, ignore it.
                // TODO: any other default handling? (eg. specific respond, trigger as a generic event, etc.);
                console.log("[Brackets LiveDev] No subscribers for message " + msg.method);
                return;
            }
        }
7663
function callback (prefix, name, message) {
      if (!message) {
        return false;
      }
      count++;
      var msg  = errMsg(prefix, name, message);
      switch(action) {
      case 'show':
        console.log(msg);
        break;
      case 'hide':
        break;
      case 'throw':
      case 'copy':
        throw new Error(msg);
      }
      return true;
    }
483466
function handle(msg) {
  log.debug('handling msg', msg)
  var to = msg.to,
    intent = msg.intent;
  if (to && intent) {
    var reply;
    // try JSON or JSON.input as input
    try {
      reply = _.get(lib_js[to], intent)(msg)
    } catch (err) {
      try {
        reply = _.get(lib_js[to], intent)(msg.input)
      } catch (deepErr) {
        log.debug('js handle fails.', deepErr)
      }
    } finally {
      // try JSON or made-JSON output
      reply = correctReply(reply, msg)
      log.debug('reply', reply)
      if (reply.to) {
        client.emit('pass', reply)
      }
    }
  }
}
1110885
function workerMessageHandler(e) {
  const msg = e.data;
  var cb;

  SHOULD_LOG && console.log("Worker received message: " + JSON.stringify(msg));

  switch (msg.event) {
    case "connInit":
      workerInitialized = true;
      workerQueue.forEach(workerInvoke);
      workerQueue = [];
      break;

    case "connStatus":
      const response = msg.data || {};
      const statusCode = response.status;
      if (connStatus[msg.conn]) {
        switch (statusCode) {
          case 200:
            connStatus[msg.conn].ready = true;
            break;
          case 401: // authorization error, need to log in
            connStatus[msg.conn].ready = false;
            connStatus[msg.conn].user = null;
            connStatus[msg.conn].anonymous = true;
            break;
          default:
            console.warn("Invalid connection response status: " + JSON.stringify(response));
            break;
        }
      }
      break;

    case "connClosed":
      cb = callbackRegistry[msg.ref];
      if (cb) {
        delete callbackRegistry[msg.ref];
        cb(msg.data);
      }
      break;

    case "connLogout":
      cb = callbackRegistry[msg.ref];
      if (cb) {
        delete callbackRegistry[msg.ref];
        cb(msg.data);
      }
      break;

    case "setState":
      const comp = componentIdx[msg.ref];
      if (comp) {
        comp.setState(msg.data);
      } else {
        SHOULD_LOG && console.warn("Component no longer registered: " + msg.ref);
      }
      break;

    case "remoteInvoke":
      // check for a callback
      cb = callbackRegistry[msg.ref];
      if (cb) {
        delete callbackRegistry[msg.ref];
        cb(msg.data);
      }
      break;

    case "login":
      // if login successful, update conn's connStatus
      if (msg.data.status === 200) {
        connStatus[msg.conn].token = msg.data.body.token;
        connStatus[msg.conn].user = msg.data.body.user;
        connStatus[msg.conn].anonymous = msg.data.body.anonymous;
      }
      // if there was a callback passed to login(), execute
      cb = callbackRegistry[msg.ref];
      if (cb) {
        delete callbackRegistry[msg.ref];
        cb(msg.data);
      }
      break;

    default:
      SHOULD_LOG && console.warn("Unreconized event from worker: " + msg.event + ". Full message: " + JSON.stringify(msg));
      break;
  }
  return;
}
556023
function messageHandler( msg ) {
  switch (msg.type) {
    case 'load'   : return handleLoadMsg(msg);
    case 'search' : return handleSearch(msg);
    case 'lookupById': return handleLookupById(msg);
    default       : logger.error('Unknown message:', msg);
  }
}
14140
def handle(self, targetdir, app=None, **options):
        
        translation.activate(settings.LANGUAGE_CODE)
        if app:
            unpack = app.split('.')

            if len(unpack) == 2:
                models = [get_model(unpack[0], unpack[1])]
            elif len(unpack) == 1:
                models = get_models(get_app(unpack[0]))
        else:
            models = get_models()
        
        messagemaker = MakeModelMessages(targetdir)
        
        for model in models:
            if hasattr(model, 'localized_fields'):
                for instance in model.objects.all():
                    messagemaker(instance)
1166967
function handleIncomingMessage(data) {
		var message;
		reader.parse(data, initPending);
		initPending = false;
		while (!reader.isEmpty()) {
			message = reader.pop();
			// console.log("IN MESSAGE: " + inspect(message));
			handlers[message.verb](message);
		}
	}
213937
function create_message(req, data) {
		debug.assert(data).is('object');
		if(!is.uuid(data.$id)) {
			data.$id = require('node-uuid').v4();
		}
		req.session.client.messages[data.$id] = data;
	}
652865
function handleMessageFromHub(ws, json, device_pubkey, bIndirectCorrespondent, callbacks){
	var subject = json.subject;
	var body = json.body;
	if (!subject || typeof body == "undefined")
		return callbacks.ifError("no subject or body");
	//if (bIndirectCorrespondent && ["cancel_new_wallet", "my_xpubkey", "new_wallet_address"].indexOf(subject) === -1)
	//    return callbacks.ifError("you're indirect correspondent, cannot trust "+subject+" from you");
	var from_address = objectHash.getDeviceAddress(device_pubkey);
	
	switch (subject){
		case "pairing":
			device.handlePairingMessage(json, device_pubkey, callbacks);
			break;
		
		case "text":
			message_counter++;
			if (!ValidationUtils.isNonemptyString(body))
				return callbacks.ifError("text body must be string");
			// the wallet should have an event handler that displays the text to the user
			eventBus.emit("text", from_address, body, message_counter);
			callbacks.ifOk();
			break;

		case "removed_paired_device":
			if(conf.bIgnoreUnpairRequests) {
				// unpairing is ignored
				callbacks.ifError("removed_paired_device ignored: "+from_address);
			} else {
				determineIfDeviceCanBeRemoved(from_address, function(bRemovable){
					if (!bRemovable)
						return callbacks.ifError("device "+from_address+" is not removable");
					device.removeCorrespondentDevice(from_address, function(){
						eventBus.emit("removed_paired_device", from_address);
						callbacks.ifOk();
					});
				});
			}
			break;

		case "chat_recording_pref":
			message_counter++;
			eventBus.emit("chat_recording_pref", from_address, body, message_counter);
			callbacks.ifOk();
			break;
		
		case "create_new_wallet":
			// {wallet: "base64", wallet_definition_template: [...]}
			walletDefinedByKeys.handleOfferToCreateNewWallet(body, from_address, callbacks);
			break;
		
		case "cancel_new_wallet":
			// {wallet: "base64"}
			if (!ValidationUtils.isNonemptyString(body.wallet))
				return callbacks.ifError("no wallet");
			walletDefinedByKeys.deleteWallet(body.wallet, from_address, callbacks.ifOk);
			break;
		
		case "my_xpubkey": // allowed from non-correspondents
			// {wallet: "base64", my_xpubkey: "base58"}
			if (!ValidationUtils.isNonemptyString(body.wallet))
				return callbacks.ifError("no wallet");
			if (!ValidationUtils.isNonemptyString(body.my_xpubkey))
				return callbacks.ifError("no my_xpubkey");
			if (body.my_xpubkey.length > 112)
				return callbacks.ifError("my_xpubkey too long");
			walletDefinedByKeys.addDeviceXPubKey(body.wallet, from_address, body.my_xpubkey, callbacks.ifOk);
			break;
		
		case "wallet_fully_approved": // allowed from non-correspondents
			// {wallet: "base64"}
			if (!ValidationUtils.isNonemptyString(body.wallet))
				return callbacks.ifError("no wallet");
			walletDefinedByKeys.handleNotificationThatWalletFullyApproved(body.wallet, from_address, callbacks.ifOk);
			break;
		
		case "new_wallet_address":
			// {wallet: "base64", is_change: (0|1), address_index: 1234, address: "BASE32"}
			if (!ValidationUtils.isNonemptyString(body.wallet))
				return callbacks.ifError("no wallet");
			if (!(body.is_change === 0 || body.is_change === 1))
				return callbacks.ifError("bad is_change");
			if (!ValidationUtils.isNonnegativeInteger(body.address_index))
				return callbacks.ifError("bad address_index");
			if (!ValidationUtils.isValidAddress(body.address))
				return callbacks.ifError("no address or bad address");
			walletDefinedByKeys.addNewAddress(body.wallet, body.is_change, body.address_index, body.address, function(err){
				if (err)
					return callbacks.ifError(err);
				callbacks.ifOk();
			});
			break;
			
		case "create_new_shared_address":
			// {address_definition_template: [...]}
			if (!ValidationUtils.isArrayOfLength(body.address_definition_template, 2))
				return callbacks.ifError("no address definition template");
			walletDefinedByAddresses.validateAddressDefinitionTemplate(
				body.address_definition_template, from_address, 
				function(err, assocMemberDeviceAddressesBySigningPaths){
					if (err)
						return callbacks.ifError(err);
					// this event should trigger a confirmatin dialog, user needs to approve creation of the shared address and choose his 
					// own address that is to become a member of the shared address
					eventBus.emit("create_new_shared_address", body.address_definition_template, assocMemberDeviceAddressesBySigningPaths);
					callbacks.ifOk();
				}
			);
			break;
		
		case "approve_new_shared_address":
			// {address_definition_template_chash: "BASE32", address: "BASE32", device_addresses_by_relative_signing_paths: {...}}
			if (!ValidationUtils.isValidAddress(body.address_definition_template_chash))
				return callbacks.ifError("invalid addr def c-hash");
			if (!ValidationUtils.isValidAddress(body.address))
				return callbacks.ifError("invalid address");
			if (typeof body.device_addresses_by_relative_signing_paths !== "object" 
					|| Object.keys(body.device_addresses_by_relative_signing_paths).length === 0)
				return callbacks.ifError("invalid device_addresses_by_relative_signing_paths");
			walletDefinedByAddresses.approvePendingSharedAddress(body.address_definition_template_chash, from_address, 
				body.address, body.device_addresses_by_relative_signing_paths);
			callbacks.ifOk();
			break;
			
		case "reject_new_shared_address":
			// {address_definition_template_chash: "BASE32"}
			if (!ValidationUtils.isValidAddress(body.address_definition_template_chash))
				return callbacks.ifError("invalid addr def c-hash");
			walletDefinedByAddresses.deletePendingSharedAddress(body.address_definition_template_chash);
			callbacks.ifOk();
			break;
			
		case "new_shared_address":
			// {address: "BASE32", definition: [...], signers: {...}}
			walletDefinedByAddresses.handleNewSharedAddress(body, {
				ifError: callbacks.ifError,
				ifOk: function(){
					callbacks.ifOk();
					eventBus.emit('maybe_new_transactions');
				}
			});
			break;
			
		// request to sign a unit created on another device
		// two use cases:
		// 1. multisig: same address hosted on several devices
		// 2. multilateral signing: different addresses signing the same message, such as a (dumb) contract
		case "sign":
			// {address: "BASE32", signing_path: "r.1.2.3", unsigned_unit: {...}}
			if (!ValidationUtils.isValidAddress(body.address))
				return callbacks.ifError("no address or bad address");
			if (!ValidationUtils.isNonemptyString(body.signing_path) || body.signing_path.charAt(0) !== 'r')
				return callbacks.ifError("bad signing path");
			var objUnit = body.unsigned_unit;
			if (typeof objUnit !== "object")
				return callbacks.ifError("no unsigned unit");
			// replace all existing signatures with placeholders so that signing requests sent to us on different stages of signing become identical,
			// hence the hashes of such unsigned units are also identical
			objUnit.authors.forEach(function(author){
				var authentifiers = author.authentifiers;
				for (var path in authentifiers)
					authentifiers[path] = authentifiers[path].replace(/./, '-'); 
			});
			var assocPrivatePayloads = body.private_payloads;
			if ("private_payloads" in body){
				if (typeof assocPrivatePayloads !== "object" || !assocPrivatePayloads)
					return callbacks.ifError("bad private payloads");
				for (var payload_hash in assocPrivatePayloads){
					var payload = assocPrivatePayloads[payload_hash];
					var hidden_payload = _.cloneDeep(payload);
					if (payload.denomination) // indivisible asset.  In this case, payload hash is calculated based on output_hash rather than address and blinding
						hidden_payload.outputs.forEach(function(o){
							delete o.address;
							delete o.blinding;
						});
					var calculated_payload_hash = objectHash.getBase64Hash(hidden_payload);
					if (payload_hash !== calculated_payload_hash)
						return callbacks.ifError("private payload hash does not match");
					if (!ValidationUtils.isNonemptyArray(objUnit.messages))
						return callbacks.ifError("no messages in unsigned unit");
					if (objUnit.messages.filter(function(objMessage){ return (objMessage.payload_hash === payload_hash); }).length !== 1)
						return callbacks.ifError("no such payload hash in the messages");
				}
			}
			// findAddress handles both types of addresses
			findAddress(body.address, body.signing_path, {
				ifError: callbacks.ifError,
				ifLocal: function(objAddress){
					// the commented check would make multilateral signing impossible
					//db.query("SELECT 1 FROM extended_pubkeys WHERE wallet=? AND device_address=?", [row.wallet, from_address], function(sender_rows){
					//    if (sender_rows.length !== 1)
					//        return callbacks.ifError("sender is not cosigner of this address");
						callbacks.ifOk();
						objUnit.unit = objectHash.getUnitHash(objUnit);
						var objJoint = {unit: objUnit, unsigned: true};
						eventBus.once("validated-"+objUnit.unit, function(bValid){
							if (!bValid){
								console.log("===== unit in signing request is invalid");
								return;
							}
							// This event should trigger a confirmation dialog.
							// If we merge coins from several addresses of the same wallet, we'll fire this event multiple times for the same unit.
							// The event handler must lock the unit before displaying a confirmation dialog, then remember user's choice and apply it to all
							// subsequent requests related to the same unit
							eventBus.emit("signing_request", objAddress, body.address, objUnit, assocPrivatePayloads, from_address, body.signing_path);
						});
						// if validation is already under way, handleOnlineJoint will quickly exit because of assocUnitsInWork.
						// as soon as the previously started validation finishes, it will trigger our event handler (as well as its own)
						network.handleOnlineJoint(ws, objJoint);
					//});
				},
				ifRemote: function(device_address){
					if (device_address === from_address){
						callbacks.ifError("looping signing request for address "+body.address+", path "+body.signing_path);
						throw Error("looping signing request for address "+body.address+", path "+body.signing_path);
					}
					var text_to_sign = objectHash.getUnitHashToSign(body.unsigned_unit).toString("base64");
					// I'm a proxy, wait for response from the actual signer and forward to the requestor
					eventBus.once("signature-"+device_address+"-"+body.address+"-"+body.signing_path+"-"+text_to_sign, function(sig){
						sendSignature(from_address, text_to_sign, sig, body.signing_path, body.address);
					});
					// forward the offer to the actual signer
					device.sendMessageToDevice(device_address, subject, body);
					callbacks.ifOk();
				},
				ifMerkle: function(bLocal){
					callbacks.ifError("there is merkle proof at signing path "+body.signing_path);
				},
				ifUnknownAddress: function(){
					callbacks.ifError("not aware of address "+body.address+" but will see if I learn about it later");
					eventBus.once("new_address-"+body.address, function(){
						// rewrite callbacks to avoid duplicate unlocking of mutex
						handleMessageFromHub(ws, json, device_pubkey, bIndirectCorrespondent, { ifOk: function(){}, ifError: function(){} });
					});
				}
			});
			break;
		
		case "signature":
			// {signed_text: "base64 of sha256", signing_path: "r.1.2.3", signature: "base64"}
			if (!ValidationUtils.isStringOfLength(body.signed_text, constants.HASH_LENGTH)) // base64 of sha256
				return callbacks.ifError("bad signed text");
			if (!ValidationUtils.isStringOfLength(body.signature, constants.SIG_LENGTH) && body.signature !== '[refused]')
				return callbacks.ifError("bad signature length");
			if (!ValidationUtils.isNonemptyString(body.signing_path) || body.signing_path.charAt(0) !== 'r')
				return callbacks.ifError("bad signing path");
			if (!ValidationUtils.isValidAddress(body.address))
				return callbacks.ifError("bad address");
			eventBus.emit("signature-" + from_address + "-" + body.address + "-" + body.signing_path + "-" + body.signed_text, body.signature);
			callbacks.ifOk();
			break;
			
		case 'private_payments':
			var arrChains = body.chains;
			if (!ValidationUtils.isNonemptyArray(arrChains))
				return callbacks.ifError("no chains found");
			profiler.increment();
			
			if (conf.bLight)
				network.requestUnfinishedPastUnitsOfPrivateChains(arrChains); // it'll work in the background
			
			var assocValidatedByKey = {};
			var bParsingComplete = false;
			var cancelAllKeys = function(){
				for (var key in assocValidatedByKey)
					eventBus.removeAllListeners(key);
			};

			var current_message_counter = ++message_counter;

			var checkIfAllValidated = function(){
				if (!assocValidatedByKey) // duplicate call - ignore
					return console.log('duplicate call of checkIfAllValidated');
				for (var key in assocValidatedByKey)
					if (!assocValidatedByKey[key])
						return console.log('not all private payments validated yet');
				assocValidatedByKey = null; // to avoid duplicate calls
				if (!body.forwarded)
					emitNewPrivatePaymentReceived(from_address, arrChains, current_message_counter);
				profiler.print();
			};
			
			async.eachSeries(
				arrChains,
				function(arrPrivateElements, cb){ // validate each chain individually
					var objHeadPrivateElement = arrPrivateElements[0];
					if (!!objHeadPrivateElement.payload.denomination !== ValidationUtils.isNonnegativeInteger(objHeadPrivateElement.output_index))
						return cb("divisibility doesn't match presence of output_index");
					var output_index = objHeadPrivateElement.payload.denomination ? objHeadPrivateElement.output_index : -1;
					var payload_hash = objectHash.getBase64Hash(objHeadPrivateElement.payload);
					var key = 'private_payment_validated-'+objHeadPrivateElement.unit+'-'+payload_hash+'-'+output_index;
					assocValidatedByKey[key] = false;
					network.handleOnlinePrivatePayment(ws, arrPrivateElements, true, {
						ifError: function(error){
							console.log("handleOnlinePrivatePayment error: "+error);
							cb("an error"); // do not leak error message to the hub
						},
						ifValidationError: function(unit, error){
							console.log("handleOnlinePrivatePayment validation error: "+error);
							cb("an error"); // do not leak error message to the hub
						},
						ifAccepted: function(unit){
							console.log("handleOnlinePrivatePayment accepted");
							assocValidatedByKey[key] = true;
							cb(); // do not leak unit info to the hub
						},
						// this is the most likely outcome for light clients
						ifQueued: function(){
							console.log("handleOnlinePrivatePayment queued, will wait for "+key);
							eventBus.once(key, function(bValid){
								if (!bValid)
									return cancelAllKeys();
								assocValidatedByKey[key] = true;
								if (bParsingComplete)
									checkIfAllValidated();
								else
									console.log('parsing incomplete yet');
							});
							cb();
						}
					});
				},
				function(err){
					bParsingComplete = true;
					if (err){
						cancelAllKeys();
						return callbacks.ifError(err);
					}
					checkIfAllValidated();
					callbacks.ifOk();
					// forward the chains to other members of output addresses
					if (!body.forwarded)
						forwardPrivateChainsToOtherMembersOfOutputAddresses(arrChains);
				}
			);
			break;
			
		case 'payment_notification':
			// note that since the payments are public, an evil user might notify us about a payment sent by someone else 
			// (we'll be fooled to believe it was sent by the evil user).  It is only possible if he learns our address, e.g. if we make it public.
			// Normally, we generate a one-time address and share it in chat session with the future payer only.
			var current_message_counter = ++message_counter;
			var unit = body;
			if (!ValidationUtils.isStringOfLength(unit, constants.HASH_LENGTH))
				return callbacks.ifError("invalid unit in payment notification");
			var bEmitted = false;
			var emitPn = function(objJoint){
				if (bEmitted)
					return;
				bEmitted = true;
				emitNewPublicPaymentReceived(from_address, objJoint.unit, current_message_counter);
			};
			eventBus.once('saved_unit-'+unit, emitPn);
			storage.readJoint(db, unit, {
				ifNotFound: function(){
					console.log("received payment notification for unit "+unit+" which is not known yet, will wait for it");
					callbacks.ifOk();
				},
				ifFound: function(objJoint){
					emitPn(objJoint);
					eventBus.removeListener('saved_unit-'+unit, emitPn);
					callbacks.ifOk();
				}
			});
			break;
			
		default:
			callbacks.ifError("unknnown subject: "+subject);
	}
}
259155
function createMessageEvent(config) {
  const { type, origin, data, target } = config;
  const messageEvent = new MessageEvent(type, {
    data,
    origin
  });

  if (target) {
    messageEvent.target = target;
    messageEvent.srcElement = target;
    messageEvent.currentTarget = target;
  }

  return messageEvent;
}
648290
def on_message(self, message):
        
        log.debug('Got websocket message %s', message)
        if message == 'websocket client ready':
            if self.connected:
                # TBD: Add ability to run commands in this branch
                log.debug('Websocket already connected, returning')
                return

            self.connected = True

            evt_processor = event_processor.SaltInfo(self)
            client = salt.netapi.NetapiClient(self.application.opts)
            client.run({
                'fun': 'grains.items',
                'tgt': '*',
                'token': self.token,
                'mode': 'client',
                'asynchronous': 'local_async',
                'client': 'local'
                })
            while True:
                try:
                    event = yield self.application.event_listener.get_event(self)
                    evt_processor.process(event, self.token, self.application.opts)
                    # self.write_message('data: {0}\n\n'.format(salt.utils.json.dumps(event, _json_module=_json)))
                except Exception as err:
                    log.debug('Error! Ending server side websocket connection. Reason = %s', err)
                    break

            self.close()
        else:
            # TBD: Add logic to run salt commands here
            pass