/*
type guard for data container send from the consumer microservice to the JM.
aka "newJobSocket" event
978371
function isJobOptProxy(data) {
    if (!data.hasOwnProperty('script') && !data.hasOwnProperty('inputs'))
        return false;
    if (!isStream(data.script)) {
        logger.error("jobOptProxy script value is not a readable stream");
        logger.error(`${data.script}`);
        return false;
    }
    for (let k in data.inputs) {
        if (!isStream(data.inputs[k])) {
            logger.error(`jobOptProxy input value \"${k}\" is not a readable stream`);
            return false;
        }
    }
    return true;
}
128969
@Override
    public void notifyOfJoin(
            int hostId, SocketChannel socket,
            SSLEngine sslEngine,
            InetSocketAddress listeningAddress,
            JSONObject jo) {
        networkLog.info(getHostId() + " notified of " + hostId);
        prepSocketChannel(socket);
        ForeignHost fhost = null;
        try {
            fhost = new ForeignHost(this, hostId, socket, m_config.deadHostTimeout,
                    listeningAddress,
                    createPicoNetwork(sslEngine, socket, false));
            putForeignHost(hostId, fhost);
            fhost.enableRead(VERBOTEN_THREADS);
        } catch (java.io.IOException e) {
            org.voltdb.VoltDB.crashLocalVoltDB("", true, e);
        }
        m_acceptor.accrue(hostId, jo);
    }
951366
private void sendLifecycleEvent(final LifecycleEvent event) {

		// instantiate and call lifecycle callbacks from configuration file
		final String listeners = Settings.LifecycleListeners.getValue();
		if (listeners != null) {

			final String[] listenerClasses = listeners.split("[\\s ,;]+");
			for (String listenerClass : listenerClasses) {

				if (StringUtils.isNotBlank(listenerClass)) {

					try {
						final HttpServiceLifecycleListener listener = (HttpServiceLifecycleListener) Class.forName(listenerClass).newInstance();
						switch (event) {

							case Started:
								listener.serverStarted();
								break;

							case Stopped:
								listener.serverStopped();
								break;
						}

					} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {

						logger.error("Unable to send lifecycle event to listener " + listenerClass, ex);
					}
				}
			}
		}
	}
246506
@Override
  public void notifyWatch(byte[] key)
  {
    int hash = _table.getPodHash(key);
    
    TablePodNode node = getNode(hash);
    
    if (node.isSelfOwner()) {
      _table.notifyWatch(key);
    }
    else {
      ServerBartender owner = node.getOwner();
      
      ClusterServiceKraken proxy = _podKraken.getProxy(owner.getId());
    
      if (proxy != null) {
        proxy.notifyWatch(_table.getKey(), key);
      }
    }
  }
1154718
function StatelessServerChannel(server, readableFactory, opts) {
  ServerChannel.call(this, server, opts);

  this._writable = undefined;
  var self = this;
  var readable;

  process.nextTick(function () {
    // Delay listening to allow handlers to be attached even if the factory is
    // purely synchronous.
    readable = readableFactory.call(self, function (err, writable) {
      process.nextTick(function () {
        // We delay once more here in case this call is synchronous, to allow
        // the readable to always be populated first.
        if (err) {
          onFinish(err);
          return;
        }
        self._writable = writable.on('finish', onFinish);
        self.emit('_writable');
      });
    }).on('data', onRequest).on('end', onEnd);
  });


  function onRequest(obj) {
    var id = obj.id;
    var buf = Buffer.concat(obj.payload);
    var err;
    try {
      var parts = readHead(HANDSHAKE_REQUEST_TYPE, buf);
      var hreq = parts.head;
      var adapter = self._getAdapter(hreq);
    } catch (cause) {
      err = toRpcError('INVALID_HANDSHAKE_REQUEST', cause);
    }

    var hres = self._createHandshakeResponse(err, hreq);
    self.emit('handshake', hreq, hres);
    if (err) {
      done(self._encodeSystemError(err));
    } else {
      self._receive(parts.tail, adapter, done);
    }

    function done(resBuf) {
      if (!self.destroyed) {
        if (!self._writable) {
          self.once('_writable', function () { done(resBuf); });
          return;
        }
        self._writable.write({
          id: id,
          payload: [HANDSHAKE_RESPONSE_TYPE.toBuffer(hres), resBuf]
        });
      }
      if (self._writable && self._endWritable) {
        self._writable.end();
      }
    }
  }

  function onEnd() { self.destroy(); }

  function onFinish(err) {
    readable
      .removeListener('data', onRequest)
      .removeListener('end', onEnd);
    self.destroy(err || true);
  }
}
1258841
protected void runEchoService(ServerSocket serverSocket) {
    if (isRunning(serverSocket)) {

      echoService = newExecutorService();

      echoService.submit(() -> {
        try {
          while (isRunning(serverSocket)) {
            Socket echoClient = serverSocket.accept();

            getLogger().info(() -> String.format("EchoClient connected from [%s]",
              echoClient.getRemoteSocketAddress()));

            echoService.submit(() -> {
              sendResponse(echoClient, receiveMessage(echoClient));
              close(echoClient);
            });
          }
        }
        catch (IOException cause) {
          if (isRunning(serverSocket)) {
            getLogger().warning(() -> String.format("An IO error occurred while listening for EchoClients:%n%s",
              ThrowableUtils.getStackTrace(cause)));
          }
        }
      });

      getLogger().info(() -> String.format("EchoServer running on port [%d]", getPort()));
    }
  }
595854
protected void sendEvent(T event, Object receiver, CreationalContext<?> creationalContext) {
        try {
            preNotify(event, receiver);
            // As we are working with the contextual instance, we may not have the
            // actual object, but a container proxy (e.g. EJB)
            notificationStrategy.invoke(receiver, observerMethod, event, beanManager, creationalContext);
        } finally {
            postNotify(event, receiver);
        }
    }
1037731
function (packetType, packetData) {
        var jobServer = this.clientOrWorker._getJobServerByUid(this.jobServerUid);
        jobServer.send(protocol.encodePacket(packetType, packetData));
	
	if(!this.clientOrWorker.readyReset)// ensure the worker won't exit
	    jobServer.send(protocol.encodePacket(protocol.PACKET_TYPES.GRAB_JOB));
	
        this.close();
    }
27257
public boolean sendInternalMessage(String worker, final String messageId, String eventName,
            final String eventMessage, boolean checkAvailableThreads) throws JMSException {
        // Spring Injection
        if (mdwMessageProducer != null) {
            mdwMessageProducer.sendMessage(eventMessage, messageId);

        }
        else {
            InternalEventDriver command = new InternalEventDriver(messageId, eventMessage);
            ThreadPoolProvider thread_pool = ApplicationContext.getThreadPoolProvider();
            if (checkAvailableThreads && !thread_pool.execute(worker, eventName, command)) {
                String msg = worker + " has no thread available for event: " + eventName
                        + " message:\n" + eventMessage;
                // make this stand out
                logger.warnException(msg, new Exception(msg));
                logger.info(thread_pool.currentStatus());
                return false;
            }
            else if (!checkAvailableThreads)
                thread_pool.execute(worker, eventName, command);

        }
        return true;
    }
1226205
private void putEvent(EventType eventType) throws Exception {
        List<EventType> eventTypes = Collections.singletonList(eventType);

        int i;
        for (i = 0; i < retryNum; ++i) {
            try {
                monitoringService.putEvents(eventTypes);
                break;
            } catch (Exception e) {
                LOG.log(Level.SEVERE, e.getMessage(), e);
            }
            Thread.sleep(retryDelay);
        }

        if (i == retryNum) {
            LOG.warning("Could not send events to monitoring service after " + retryNum + " retries.");
            throw new Exception("Send SERVER_START/SERVER_STOP event to SAM Server failed");
        }

    }
700702
private void publishPurgeEvent(WSJobInstance jobInstance) {
        if (eventsPublisher != null) {
            eventsPublisher.publishJobInstanceEvent(jobInstance, BatchEventsPublisher.TOPIC_INSTANCE_PURGED, null);
        }
    }
1355024
private static Interceptor<ByteBuf, ByteBuf> sendHello() {
        return in -> newConnection -> newConnection.writeString(Observable.just("Hello\n"))
                                                   .concatWith(in.handle(newConnection));
    }
625311
private void postRoutedServerNotificationListenerRegistrationEvent(String operation, NotificationTargetInformation nti, ObjectName listener,
                                                                       NotificationFilter filter, Object handback) {
        Map<String, Object> props = createServerListenerRegistrationEvent(operation, nti, listener, filter, handback);
        safePostEvent(new Event(REGISTER_JMX_NOTIFICATION_LISTENER_TOPIC, props));
    }
1299263
function wrapNodeSocket(socket, reportError) {
  // Data traveling from node socket to consumer
  var incoming = makeChannel();
  // Data traveling from consumer to node socket
  var outgoing = makeChannel();

  var onDrain2 = wrapHandler(onReadable, onError);
  onTake = wrapHandler(onTake, onError);

  var paused = false;
  var ended = false;

  outgoing.take(onTake);
  socket.on("error", onError);
  socket.on("drain", wrap(onDrain, onError));
  socket.on("readable", wrap(onReadable, onError));
  socket.on("end", wrap(onEnd, onError));

  function onError(err) {
    if (err.code === "EPIPE" || err.code === "ECONNRESET") {
      return onEnd();
    }
    if (reportError) return reportError(err);
    console.error(err.stack);
  }

  function onTake(data) {
    if (ended) return;
    // console.log("TAKE", data);
    if (data === undefined) {
      ended = true;
      return socket.end();
    }
    if (socket.write(data)) {
      outgoing.take(onTake);
    }
    else {
      if (!paused) paused = true;
    }
  }

  function onDrain() {
    if (paused) {
      paused = false;
      outgoing.take(onTake);
    }
  }

  function onEnd() {
    if (ended) return;
    ended = true;
    incoming.put(undefined);
  }

  function onReadable() {
    if (ended) return;
    while (true) {
      var chunk = socket.read();
      // console.log("READ", chunk);
      if (!chunk) return;
      if (!incoming.put(chunk)) {
        incoming.drain(onDrain2);
        return;
      }
    }
  }

  return {
    put: outgoing.put,
    drain: outgoing.drain,
    take: incoming.take,
  };
}
1166403
private void publishEvent(WSJobExecution objectToPublish, String eventToPublishTo) {
        if (getBatchEventsPublisher() != null) {
            getBatchEventsPublisher().publishJobExecutionEvent(objectToPublish, eventToPublishTo, correlationId);
        }
    }
864104
public synchronized void send(String data, String event, String id, EventCallback callback) {
        if (open == 0 || shutdown) {
            if (callback != null) {
                callback.failed(this, data, event, id, new ClosedChannelException());
            }
            return;
        }
        queue.add(new SSEData(event, data, id, callback));
        sink.getIoThread().execute(new Runnable() {
            @Override
            public void run() {
                synchronized (ServerSentEventConnection.this) {
                    if (pooled == null) {
                        fillBuffer();
                        writeListener.handleEvent(sink);
                    }
                }
            }
        });
    }
456867
def _domain_event_job_completed_cb(conn, domain, params, opaque):
    '''
    
    '''
    _salt_send_domain_event(opaque, conn, domain, opaque['event'], {
        'params': params
    })