-------------------------------------------------- STREAMING AND POD DATA
179014
function(url, next, headers, options) {
    var headerStruct = {
      'User-Agent': 'request'
    };

    var params = {
      url : url,
      method : 'GET'
    };

    if (headers) {
      for (var k in headers) {
        if (headers.hasOwnProperty(k)) {
          headerStruct[k] = headers[k];
        }
      }
    }

    params.headers = headerStruct;
    params.gzip = true;

    if (options) {
      for (var k in options) {
        if (options.hasOwnProperty(k)) {
          params[k] = options[k];
        }
      }
    }

    if (next) {
      request(params, function(error, res, body) {
          if (res && res.headers && -1 !== res.headers['content-type'].indexOf('json') || -1 !== res.headers['content-type'].indexOf('javascript')) {
            try {
              body = JSON.parse(body);
            } catch (e) {
              error = e.message;
            }
          }

          if (404 === res.statusCode) {
            next('Not Found', body, res.headers, res.statusCode);
          } else {
            next(error, body, res ? res.headers : null, res ? res.statusCode : null);
          }
        }
      );
    } else {
      return request(params);
    }
  }
1144503
private void broadcast() {
    try (LongStream events = eventStream()) {
      LongArrayList batch = new LongArrayList(batchSize);
      for (PrimitiveIterator.OfLong i = events.iterator(); i.hasNext();) {
        batch.add(i.nextLong());
        if (batch.size() == batchSize) {
          router.route(batch, self());
          batch = new LongArrayList(batchSize);
        }
      }
      router.route(batch, self());
      router.route(FINISH, self());
    } catch (Exception e) {
      context().system().log().error(e, "");
      context().stop(self());
    }
  }
67500
function broadcast(connections, packet) {
    var i = connections.length;
    while (i--) {
        connections[i].write(packet);
        flush(connections[i]);
    }
}
335821
public void postClassLoaderInit()
  {
    // bartender.setLocalShard(pod.getNode(podNode));
    ServiceManagerBuilderAmp builder = ServicesAmp.newManager();
    builder.name(EnvLoader.getEnvironmentName());
    
    BartenderSystem bartender = BartenderSystem.current();
    if (bartender != null) {
      PodBartender pod = bartender.findActivePod(getPodName());
    
      Objects.requireNonNull(pod);
    
      int podNode = _builder.getPodNode();
      bartender.setLocalPod(pod);
      //builder.setPodNode(new ServiceNodeImpl(pod.getNode(podNode)));
    }
    
    /*
    //builder.setJournalMaxCount(_builder.getJournalMaxCount());
    builder.setJournalDelay(_builder.getJournalDelay());
    // XXX: config timing issue
    builder.debug(_builder.isDebug());
    builder.debugQueryTimeout(_builder.getDebugQueryTimeout());
    builder.autoStart(false);
    */
    _ampManager = builder.start();

    /*
    if (_controller.getConfigException() != null) {
      setAmpConfigException(_controller.getConfigException());
    }
    else if (_configException != null) {
      setAmpConfigException(_configException);
    }
    */
   
    Amp.contextManager(_ampManager);
    
    /*
    _ampManager.run(()->{
      EmbedBuilder embedBuilder = new EmbedBuilder();
      embedBuilder.scanClassLoader();
      embedBuilder.build();
    });
    */
  }
336295
public void send(OutputStream os, 
                   HeadersAmp headers,
                   String address, 
                   String methodName,
                   PodRef podCaller,
                   Object []args)
    throws IOException
  {
    init(os);

    OutH3 out = _out;

    if (out == null) {
      return;
    }
    
    if (log.isLoggable(Level.FINEST)) {
      log.finest("hamp-send-w " + methodName + Arrays.asList(args)
                + "\n  {to:" + address + "}");
    }
  
    try {
      out.writeLong(MessageTypeHamp.SEND.ordinal());
      writeHeaders(out, headers);
    
      writeMethod(out, address, methodName, podCaller);
      writeArgs(out, args);

      // XXX: out.flushBuffer();
    } catch (Throwable e) {
      log.log(Level.WARNING, e.toString(), e);
      
      throw e;
    }
  }
620969
protected boolean podRunning(Json podStatus) {
        if(podStatus == null) {
            return false;
        }
        
        /*
         * A pod can only be considered 'running' if the following conditions are all true:
         * 1. status.phase == "Running",
         * 2. status.message is Undefined (does not exist)
         * 3. status.reason is Undefined (does not exist)
         * 4. all of status.containerStatuses[*].ready == true
         * 5. for conditions[*].type == "Ready" conditions[*].status must be "True" 
         */
        // walk through each condition step by step
        // 1 status.phase
        log.trace("Determining pod status");
        String phase = Optional.ofNullable(podStatus.at("phase"))
                .map(Json::asString)
                .orElse("not running");
        log.trace("  status.phase=%s", phase);
        if(!phase.equalsIgnoreCase("Running")) {
            return false;
        }
        // 2. and 3. status.message and status.reason
        String statusMessage = Optional.ofNullable(podStatus.at("message"))
                .map(Json::asString)
                .orElse(null);
        String statusReason = Optional.ofNullable(podStatus.at("reason"))
                .map(Json::asString)
                .orElse(null);
        log.trace("  status.message=%s and status.reason=%s", statusMessage, statusReason);
        if(statusMessage != null || statusReason != null) {
            return false;
        }
        // 4. status.containerStatuses.ready
        List<Json> containerStatuses = Optional.ofNullable(podStatus.at("containerStatuses"))
                .map(Json::asJsonList)
                .orElse(Collections.emptyList());
        boolean ready = true;
        // if we have no containerStatuses, we don't check for it and consider this condition as passed
        for(Json containerStatus: containerStatuses) {
            ready = ready && containerStatus.at("ready").asBoolean();
        }
        log.trace("  containerStatuses[].status of all container is %s", Boolean.toString(ready));
        if(!ready) {
            return false;
        }
        // 5. ready condition must be "True"
        Boolean readyCondition = Boolean.FALSE;
        List<Json> conditions = podStatus.at("conditions").asJsonList();
        // walk through all the conditions and find type=="Ready" and get the value of the status property
        for(Json condition: conditions) {
            String type = condition.at("type").asString();
            if(type.equalsIgnoreCase("Ready")) {
                readyCondition = new Boolean(condition.at("status").asString());
            }
        }
        log.trace(  "conditions with type==\"Ready\" has status property value = %s", readyCondition.toString());
        if(!readyCondition.booleanValue()) {
            return false;
        }
        return true;
    }
1023243
void writeSnapshotData(DataOutputView out) throws IOException {
		writeTypeClass(out);
		writeKryoRegistrations(out, kryoRegistrations);
		writeDefaultKryoSerializers(out, defaultKryoSerializers);
		writeDefaultKryoSerializerClasses(out, defaultKryoSerializerClasses);
	}
143770
@Override
	public void broadcast(String name, Tree payload, Groups groups, boolean local) {
		String key = getCacheKey(name, groups);
		ListenerEndpoint[] endpoints;
		if (local) {
			endpoints = localBroadcasterCache.get(key);
		} else {
			endpoints = broadcasterCache.get(key);
		}
		if (endpoints == null) {
			HashSet<ListenerEndpoint> list = new HashSet<>();
			readLock.lock();
			try {
				for (Map.Entry<String, HashMap<String, Strategy<ListenerEndpoint>>> entry : listeners.entrySet()) {
					if (Matcher.matches(name, entry.getKey())) {
						for (Map.Entry<String, Strategy<ListenerEndpoint>> test : entry.getValue().entrySet()) {
							if (groups != null) {
								final String testGroup = test.getKey();
								for (String group : groups.groups()) {
									if (group.equals(testGroup)) {
										for (ListenerEndpoint endpoint : test.getValue().getAllEndpoints()) {
											if (local) {
												if (endpoint.isLocal()) {
													list.add(endpoint);
												}
											} else {
												list.add(endpoint);
											}
										}
									}
								}
							} else {
								if (local) {
									for (ListenerEndpoint endpoint : test.getValue().getAllEndpoints()) {
										if (local) {
											list.add(endpoint);
										}
									}
								} else {
									list.addAll(test.getValue().getAllEndpoints());
								}
							}
						}
					}
				}
			} finally {
				readLock.unlock();
			}
			endpoints = new ListenerEndpoint[list.size()];
			list.toArray(endpoints);
			if (local) {
				localBroadcasterCache.put(key, endpoints);
			} else {
				broadcasterCache.put(key, endpoints);
			}
		}
		if (endpoints.length == 0) {
			return;
		}
		if (endpoints.length == 1) {
			try {
				endpoints[0].on(name, payload, groups, true);
			} catch (Exception cause) {
				logger.error("Unable to invoke event listener!", cause);
			}
			return;
		}
		HashSet<String> nodeSet = new HashSet<>(endpoints.length * 2);
		for (ListenerEndpoint endpoint : endpoints) {
			if (endpoint.isLocal() || nodeSet.add(endpoint.getNodeID())) {
				try {
					endpoint.on(name, payload, groups, true);
				} catch (Exception cause) {
					logger.error("Unable to invoke event listener!", cause);
				}
			}
		}
	}
358533
function (frameData) {
      // Would be better to check controller.streaming() in showOverlay, but that method doesn't exist, yet.
      this.setGraphic('wave');
      if (frameData.hands.length > 0) {
        this.recording.addFrame(frameData);
        this.hideOverlay();
      } else if ( !this.recording.blank() ) {
        this.finishRecording();
      }
    }
537320
function() {
                if (hasCache) {
                    queueDb = new Dexie('podQueues');

                    queueDb.version(1)
                        .stores({
                            waitQ: '[owner+threadId+uniqueId], owner, threadId, uniqueId, message'
                        });

                    if (enableCache) {
                        db = new Dexie('podChat');

                        db.version(1)
                            .stores({
                                users: '&id, name, cellphoneNumber, keyId',
                                contacts: '[owner+id], id, owner, uniqueId, userId, cellphoneNumber, email, firstName, lastName, expireTime',
                                threads: '[owner+id] ,id, owner, title, time, [owner+time]',
                                participants: '[owner+id], id, owner, threadId, notSeenDuration, name, contactName, email, expireTime',
                                messages: '[owner+id], id, owner, threadId, time, [threadId+id], [threadId+owner+time]',
                                messageGaps: '[owner+id], [owner+waitsFor], id, waitsFor, owner, threadId, time, [threadId+owner+time]'
                            });

                        db.open()
                            .catch(function(e) {
                                console.log('Open failed: ' + e.stack);
                            });

                        db.on('ready', function() {
                            isCacheReady = true;
                        }, true);

                        db.on('versionchange', function(event) {
                            window.location.reload();
                        });
                    }
                }
                else {
                    console.log(CHAT_ERRORS[6600]);
                }
            }
258437
function(data) {
    if(!data || !data.sys) {
      return;
    }
    dict = data.sys.dict;
    var protos = data.sys.protos;

    //Init compress dict
    if(dict) {
      dict = dict;
      abbrs = {};

      for(var route in dict) {
        abbrs[dict[route]] = route;
      }
    }

    //Init protobuf protos
    if(protos) {
      protoVersion = protos.version || 0;
      serverProtos = protos.server || {};
      clientProtos = protos.client || {};

        //Save protobuf protos to localStorage
        window.localStorage.setItem('protos', JSON.stringify(protos));

        if(!!protobuf) {
          protobuf.init({encoderProtos: protos.client, decoderProtos: protos.server});
        }
        if(!!decodeIO_protobuf) {
          decodeIO_encoder = decodeIO_protobuf.loadJson(clientProtos);
          decodeIO_decoder = decodeIO_protobuf.loadJson(serverProtos);
        }
      }
    }
599649
private void loadMoreData() {
        isLoading = true;
        Util.getClient()
                .getCloudItemsAsync(sourceId, currentPath, nextTokens.get(currentPath))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(this);
    }
1063391
@Override
  public void offerData(OutputStream stream) {
    byte[] next = null;
    try {
      next = queue.poll(timeoutMillis, TimeUnit.MILLISECONDS);
    } catch (InterruptedException ignored) {
      // If this ever times out (unlikely given our assumption), next stays null.
    }

    try {
      // Only if next is not null nor CLOSED do we write the chunk.
      if (next != null && next != CLOSED) {
        stream.write(next);
        return;
      }
    } catch (IOException ignored) {
      // Any errors in the piped (NS)OutputStream (unlikely) fall through to the next section.
    }

    // Close the piped (NS)OutputStream and ignore any errors.
    try {
      stream.close();
    } catch (IOException ignored) {
      // Ignored.
    }
  }
167958
public function main()
    {
        declare(ticks = 1);
        pcntl_signal(SIGINT, function () {
            exit(0);
        });

        $kubernetesClient = $this->getKubernetesClient();
        $storeEnabled = $this->getStoreEnabled();

        if ($storeEnabled) {
            $this->store = new Store($this);
            $this->store->init();
        }

        $watches = new \KubernetesClient\WatchCollection();

        $configMapNamespace = $this->getConfigMapNamespace();
        $configMapName = $this->getConfigMapName();

        $storeNamespace = $this->getStoreNamespace();
        $storeName = $this->getStoreName();

        //controller config
        $watch = $kubernetesClient->createWatch("/api/v1/watch/namespaces/${configMapNamespace}/configmaps/${configMapName}", [], $this->getConfigMapWatchCallback());
        $watches->addWatch($watch);

        while (true) {
            usleep(100 * 1000);
            $watches->start(1);

            // do NOT perform anything until config is loaded
            if (!$this->getConfigLoaded()) {
                $this->log("waiting for ConfigMap ${configMapNamespace}/${configMapName} to be present and valid");
                sleep(5);
                continue;
            }

            // do NOT perform anything until store is ready
            if ($storeEnabled) {
                if (!$this->store->initialized()) {
                    $this->log("waiting for store ConfigMap ${storeNamespace}/${storeName} to be present and valid");
                    $this->store->init();
                    sleep(5);
                    continue;
                }
            }

            // update store data
            if ($storeEnabled) {
                $this->store->readWatches();
            }

            // process plugins
            foreach ($this->plugins as $plugin) {
                $plugin->preReadWatches();
                $plugin->readWatches();
                $plugin->postReadWatches();

                if ($plugin->getActionRequired()) {
                    // be patient in failure scenarios
                    if (!$plugin->getLastActionSuccess() && ((time() - $this->failedActionWaitTime) <= $plugin->getLastActionAttemptTime())) {
                        continue;
                    }

                    // wait for settle time
                    $settleTime = $plugin->getSettleTime();
                    $actionRequiredTime = $plugin->getActionRequiredTime();
                    if ($settleTime > 0 &&
                        $actionRequiredTime > 0 &&
                        ((time() - $actionRequiredTime) <= $settleTime)) {
                        continue;
                    }

                    // wait for throttle time
                    $throttleTime = $plugin->getThrottleTime();
                    $lastActionAttemptTime = $plugin->getLastActionAttemptTime();
                    if ($throttleTime > 0 &&
                        $lastActionAttemptTime > 0 &&
                        ((time() - $lastActionAttemptTime) <= $throttleTime)) {
                        continue;
                    }

                    $plugin->invokeAction();
                }
            }

            // update store data
            if ($storeEnabled) {
                $this->store->readWatches();
            }
        }
    }
597035
public void query(WebSocket session,
                    HeadersAmp headers,
                    String from,
                    long qid,
                    String to, 
                    String methodName,
                    PodRef podCaller,
                    Object ...args)
    throws IOException
  {
    /*
    try (Writer out = session.getBasicRemote().getSendWriter()) {
      init(out);
      
      query(headers, from, qid, to, methodName, args);
    }
    */
  }
127452
public function stream()
    {
        $patch = $this->jsonPatch->generate($this->getPath(), JsonPatch::ACTION_ADD, $this->data, true);
        $this->dataSource->save($this->process, [$patch]);
        $this->data = [];
    }
1026963
@Override
	public void streamBundleTo(String bundlePath, OutputStream out) throws ResourceNotFoundException {

		try {
			// byte[] gzip = gzipCache.get(bundlePath);
			byte[] gzip = (byte[]) cacheMgr.get(ZIP_CACHE_PREFIX + bundlePath);
			// If it's not cached yet
			if (null == gzip) {
				// Stream the stored data
				ByteArrayOutputStream baOs = new ByteArrayOutputStream();
				BufferedOutputStream bfOs = new BufferedOutputStream(baOs);
				rsHandler.streamBundleTo(bundlePath, bfOs);

				// Copy the data into the ByteBuffer
				bfOs.close();
				gzip = baOs.toByteArray();

				// Cache the byte array
				cacheMgr.put(ZIP_CACHE_PREFIX + bundlePath, gzip);
			}

			// Write bytes to the outputstream
			IOUtils.write(gzip, out);

		} catch (IOException e) {
			throw new BundlingProcessException("Unexpected IOException writing bundle[" + bundlePath + "]", e);
		}

	}