Stateless intermediate operations from Stream
771323
@Override
    public Stream<P_OUT> unordered() {
        if (!isOrdered())
            return this;
        return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_ORDERED) {
            @Override
            public Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
                return sink;
            }
        };
    }
811551
public TransitionableState getTransitionableState(final Flow flow, final String stateId) {
        if (containsFlowState(flow, stateId)) {
            return (TransitionableState) flow.getTransitionableState(stateId);
        }
        return null;
    }
647756
function recreate(makeStream, options) {
  if (options === undefined && isObject(makeStream)) {
    options = makeStream;
    makeStream = options.makeStream;
  }

  options = options || {};

  if (options.connect === undefined) {
    options.connect = true;
  }

  var connectedEvent = options.connectedEvent || 'open';

  var stream = through();
  stream.inverse = through();

  stream.inverse.pause();

  stream.write = function(chunk) {
    return stream.inverse.write(chunk);
  };

  stream.connect = function() {
    stream.underlying = makeStream();

    stream.inverse.pipe(stream.underlying, {end: false});

    stream.underlying.on('data', function(chunk) {
      return stream.emit('data', chunk);
    });

    stream.underlying.on(connectedEvent, function() {
      stream.backoff.reset();
      stream.inverse.resume();
      stream.emit('open');
      stream.emit('underlying-open');
    });

    stream.underlying.on('end', function() {
      stream.inverse.pause();
      if (!stream.preventBackoff) {
        stream.backoff.backoff();
      }
      stream.emit('underlying-end');
    });

    stream.underlying.on('error', function(e) {
      stream.inverse.pause();
      if (!stream.preventBackoff) {
        stream.backoff.backoff();
      }
      stream.emit('underlying-error');
    });
  };

  var origEnd = stream.end.bind(stream);

  stream.end = function() {
    stream.preventBackoff = true;
    stream.underlying.end();
    return origEnd();
  };

  stream.backoff = options.backoff || backoff.exponential(options);

  stream.backoff.on('backoff', function(num, delay) {
    stream.emit('backoff-schedule', num, delay);
  });

  stream.backoff.on('ready', function() {
    stream.connect();
    stream.emit('backoff-ready');
  });

  stream.backoff.on('fail', function() {
    stream.emit('backoff-fail');
  });


  if (options.connect) {
    stream.connect();
  }

  return stream;
}
1083638
private NettyClientStream.TransportState clientStream(Http2Stream stream) {
    return stream == null ? null : (NettyClientStream.TransportState) stream.getProperty(streamKey);
  }
496707
public <U> BaseSimpleReactStream<U> fromStream(final Stream<CompletableFuture<U>> stream) {

        final Stream s = stream;
        return construct(s);
    }
1334912
private static Observable<Map<String, Object>> tombstone(Observable<Map<String, Object>> instanceStream, InstanceKey instanceKey) {
        return instanceStream.publish(is -> {
            Observable<Map<String, Object>> tombstone = is
                    // collect all unique "TypeAndName" keys
                    .collect(() -> new HashSet<TypeAndNameKey>(), (listOfTypeAndName, event) -> {
                        listOfTypeAndName.add((TypeAndNameKey) event.get("TypeAndName"));
                    })
                    // when instanceStream completes, emit a "tombstone" for each "TypeAndName" in the HashSet collected above
                    .flatMap(listOfTypeAndName -> {
                        return Observable.from(listOfTypeAndName)
                                .map(typeAndName -> {
                                    Map<String, Object> tombstoneForTypeAndName = new LinkedHashMap<>();
                                    tombstoneForTypeAndName.put("tombstone", "true");
                                    tombstoneForTypeAndName.put("InstanceKey", instanceKey);
                                    tombstoneForTypeAndName.put("TypeAndName", typeAndName);
                                    tombstoneForTypeAndName.put("name", typeAndName.getName());
                                    tombstoneForTypeAndName.put("type", typeAndName.getType());
                                    return tombstoneForTypeAndName;
                                });
                    });
            
            // concat the tombstone events to the end of the original stream
            return is.mergeWith(tombstone);
        });
    }
368440
public static <T> CompletableFuture<List<T>> collectFrom(Stream<T> stream, Executor executor) {
        requireNonNull(stream, "stream");
        requireNonNull(executor, "executor");
        final CompletableFuture<List<T>> future = new CompletableFuture<>();
        executor.execute(() -> {
            try {
                future.complete(stream.collect(toImmutableList()));
            } catch (Exception e) {
                future.completeExceptionally(e);
            }
        });
        return future;
    }
345329
@PublicEvolving
	public QueryableStateStream<KEY, T> asQueryableState(
			String queryableStateName,
			ValueStateDescriptor<T> stateDescriptor) {

		transform("Queryable state: " + queryableStateName,
				getType(),
				new QueryableValueStateOperator<>(queryableStateName, stateDescriptor));

		stateDescriptor.initializeSerializerUnlessSet(getExecutionConfig());

		return new QueryableStateStream<>(
				queryableStateName,
				stateDescriptor,
				getKeyType().createSerializer(getExecutionConfig()));
	}
688492
public static <T> CloseableIterable<T> fromStream(Stream<T> stream) {
        return wrap(stream::iterator, stream);
    }
1199757
function tokenBase(stream, state) {
    var ch = stream.next(), 
        mightBeFunction = false,
        isEQName = isEQNameAhead(stream);
    
    // an XML tag (if not in some sub, chained tokenizer)
    if (ch == "<") {
      if(stream.match("!--", true))
        return chain(stream, state, tokenXMLComment);
        
      if(stream.match("![CDATA", false)) {
        state.tokenize = tokenCDATA;
        return ret("tag", "tag");
      }
      
      if(stream.match("?", false)) {
        return chain(stream, state, tokenPreProcessing);
      }
      
      var isclose = stream.eat("/");
      stream.eatSpace();
      var tagName = "", c;
      while ((c = stream.eat(/[^\s\u00a0=<>\"\'\/?]/))) tagName += c;
      
      return chain(stream, state, tokenTag(tagName, isclose));
    }
    // start code block
    else if(ch == "{") {
      pushStateStack(state,{ type: "codeblock"});
      return ret("", "");
    }
    // end code block
    else if(ch == "}") {
      popStateStack(state);
      return ret("", "");
    }
    // if we're in an XML block
    else if(isInXmlBlock(state)) {
      if(ch == ">")
        return ret("tag", "tag");
      else if(ch == "/" && stream.eat(">")) {
        popStateStack(state);
        return ret("tag", "tag");
      }
      else  
        return ret("word", "variable");
    }
    // if a number
    else if (/\d/.test(ch)) {
      stream.match(/^\d*(?:\.\d*)?(?:E[+\-]?\d+)?/);
      return ret("number", "atom");
    }
    // comment start
    else if (ch === "(" && stream.eat(":")) {
      pushStateStack(state, { type: "comment"});
      return chain(stream, state, tokenComment);
    }
    // quoted string
    else if (  !isEQName && (ch === '"' || ch === "'"))
      return chain(stream, state, tokenString(ch));
    // variable
    else if(ch === "$") {
      return chain(stream, state, tokenVariable);
    }
    // assignment
    else if(ch ===":" && stream.eat("=")) {
      return ret("operator", "keyword");
    }
    // open paren
    else if(ch === "(") {
      pushStateStack(state, { type: "paren"});
      return ret("", "");
    }
    // close paren
    else if(ch === ")") {
      popStateStack(state);
      return ret("", "");
    }
    // open paren
    else if(ch === "[") {
      pushStateStack(state, { type: "bracket"});
      return ret("", "");
    }
    // close paren
    else if(ch === "]") {
      popStateStack(state);
      return ret("", "");
    }
    else {
      var known = keywords.propertyIsEnumerable(ch) && keywords[ch];

      // if there's a EQName ahead, consume the rest of the string portion, it's likely a function
      if(isEQName && ch === '\"') while(stream.next() !== '"'){}
      if(isEQName && ch === '\'') while(stream.next() !== '\''){}
      
      // gobble up a word if the character is not known
      if(!known) stream.eatWhile(/[\w\$_-]/);
      
      // gobble a colon in the case that is a lib func type call fn:doc
      var foundColon = stream.eat(":");
      
      // if there's not a second colon, gobble another word. Otherwise, it's probably an axis specifier
      // which should get matched as a keyword
      if(!stream.eat(":") && foundColon) {
        stream.eatWhile(/[\w\$_-]/);
      }
      // if the next non whitespace character is an open paren, this is probably a function (if not a keyword of other sort)
      if(stream.match(/^[ \t]*\(/, false)) {
        mightBeFunction = true;
      }
      // is the word a keyword?
      var word = stream.current();
      known = keywords.propertyIsEnumerable(word) && keywords[word];
      
      // if we think it's a function call but not yet known, 
      // set style to variable for now for lack of something better
      if(mightBeFunction && !known) known = {type: "function_call", style: "variable def"};
      
      // if the previous word was element, attribute, axis specifier, this word should be the name of that
      if(isInXmlConstructor(state)) {
        popStateStack(state);
        return ret("word", "variable", word);
      }
      // as previously checked, if the word is element,attribute, axis specifier, call it an "xmlconstructor" and 
      // push the stack so we know to look for it on the next word
      if(word == "element" || word == "attribute" || known.type == "axis_specifier") pushStateStack(state, {type: "xmlconstructor"});
      
      // if the word is known, return the details of that else just call this a generic 'word'
      return known ? ret(known.type, known.style, word) :
                     ret("word", "variable", word);
    }
  }
383673
@Override
    public AnyM<W, ? extends ReactiveSeq<T>> iterate(UnaryOperator<T> fn, T alt) {

        return super.iterate(fn,alt);
    }
2367
function (stream, state) {
        var plannedToken = state.plannedTokens.shift();
        if (plannedToken === undefined) {
            return null;
        }
        stream.match(plannedToken.value);
        var tokens = plannedToken.type.split('.');
        return cmTokenFromAceTokens(tokens);
    }
1250572
def FromStream(cls, stream, auto_transfer=True, total_size=None, **kwds):
        """"""
        return cls(stream, auto_transfer=auto_transfer, total_size=total_size,
                   **kwds)
909494
function (locals) {
    var stream = new ReadableStream();
    
    // streaming will be set to false whenever there is back-pressure
    var streaming = false;

    function release() {
      streaming = true;
    }

    // make sure _read is always implemented
    stream._read = release;

    // then-yield unwrap function
    // which implements the backpressure pause mechanism
    function unwrap(value) {
      if (streaming) return value;
      return new Promise(function (resolve) {
        stream._read = function () {
          release();
          this._read = release;
          resolve(value);
        }
      });
    }

    var template = fn(locals, runtime, {
      push: function () {
        for (var i = 0; i < arguments.length; i++)
          if (!stream.push(arguments[i].toString())) streaming = false;
      }
    });

    // call our function, setting `streaming` to `false` whenever
    // the buffer is full and there is back-pressure
    var result = ty.spawn(template, Promise.resolve, unwrap);

    // once the function completes, we end the stream by pushing `null`
    if (result)
      result.then(stream.push.bind(stream, null), function (err) {
        stream.emit('error', err);
        stream.push(null);
      });
    else
      stream.push(null);

    return stream;
  }
1122387
@SafeVarargs
    @SuppressWarnings("unchecked")
    public static <T> Stream<T> stream(Iterable<T>... iterables) {
        List<Iterator<T>> iterators = Arrays.stream(iterables)
            .map(Iterable::iterator)
            .collect(Collectors.toList());

        return stream(iterators.toArray(new Iterator[iterables.length]));
    }
879353
function to_explicit(stream, state) {
    var token = null;

    if (phase(state) == rx_substitution ||
        stream.match(rx_substitution, false)) {

      switch (stage(state)) {
      case 0:
        change(state, to_explicit, context(rx_substitution, 1));
        assert(stream.match(rx_substitution_text));
        token = 'variable-2';
        break;
      case 1:
        change(state, to_explicit, context(rx_substitution, 2));
        assert(stream.match(rx_substitution_sepa));
        break;
      case 2:
        change(state, to_explicit, context(rx_substitution, 3));
        assert(stream.match(rx_substitution_name));
        token = 'keyword';
        break;
      case 3:
        change(state, to_explicit, context(rx_substitution, 4));
        assert(stream.match(rx_substitution_tail));
        token = 'meta';
        break;
      default:
        change(state, to_normal);
        assert(stream.current() == '');
      }
    } else if (phase(state) == rx_directive ||
               stream.match(rx_directive, false)) {

      switch (stage(state)) {
      case 0:
        change(state, to_explicit, context(rx_directive, 1));
        assert(stream.match(rx_directive_name));
        token = 'keyword';

        if (stream.current().match(/^(?:math|latex)/))
          state.tmp_stex = true;
        else if (stream.current().match(/^python/))
          state.tmp_py = true;
        break;
      case 1:
        change(state, to_explicit, context(rx_directive, 2));
        assert(stream.match(rx_directive_tail));
        token = 'meta';

        if (stream.match(/^latex\s*$/) || state.tmp_stex) {
          state.tmp_stex = undefined; change(state, to_mode, {
            mode: mode_stex, local: CodeMirror.startState(mode_stex)
          });
        }
        break;
      case 2:
        change(state, to_explicit, context(rx_directive, 3));
        if (stream.match(/^python\s*$/) || state.tmp_py) {
          state.tmp_py = undefined; change(state, to_mode, {
            mode: mode_python, local: CodeMirror.startState(mode_python)
          });
        }
        break;
      default:
        change(state, to_normal);
        assert(stream.current() == '');
      }
    } else if (phase(state) == rx_link || stream.match(rx_link, false)) {

      switch (stage(state)) {
      case 0:
        change(state, to_explicit, context(rx_link, 1));
        assert(stream.match(rx_link_head));
        assert(stream.match(rx_link_name));
        token = 'link';
        break;
      case 1:
        change(state, to_explicit, context(rx_link, 2));
        assert(stream.match(rx_link_tail));
        token = 'meta';
        break;
      default:
        change(state, to_normal);
        assert(stream.current() == '');
      }
    } else if (stream.match(rx_footnote)) {
      change(state, to_normal);
      token = 'quote';
    } else if (stream.match(rx_citation)) {
      change(state, to_normal);
      token = 'quote';
    }

    else {
      stream.eatSpace();
      if (stream.eol()) {
        change(state, to_normal);
      } else {
        stream.skipToEnd();
        change(state, to_comment);
        token = 'comment';
      }
    }

    return token;
  }
727732
private <K, V, T extends RedisChannelHandler<K, V>, S> ConnectionFuture<S> connectStatefulAsync(T connection,
            RedisCodec<K, V> codec,
            DefaultEndpoint endpoint, RedisURI connectionSettings, Mono<SocketAddress> socketAddressSupplier,
            Supplier<CommandHandler> commandHandlerSupplier) {

        ConnectionBuilder connectionBuilder = createConnectionBuilder(connection, endpoint, connectionSettings,
                socketAddressSupplier, commandHandlerSupplier);

        if (clientOptions.isPingBeforeActivateConnection()) {
            if (hasPassword(connectionSettings)) {
                connectionBuilder.enableAuthPingBeforeConnect();
            } else {
                connectionBuilder.enablePingBeforeConnect();
            }
        }

        ConnectionFuture<RedisChannelHandler<K, V>> future = initializeChannelAsync(connectionBuilder);
        ConnectionFuture<?> sync = future;

        if (!clientOptions.isPingBeforeActivateConnection() && hasPassword(connectionSettings)) {

            sync = sync.thenCompose(channelHandler -> {

                CommandArgs<K, V> args = new CommandArgs<>(codec).add(connectionSettings.getPassword());
                AsyncCommand<K, V, String> command = new AsyncCommand<>(new Command<>(CommandType.AUTH, new StatusOutput<>(
                        codec), args));

                if (connection instanceof StatefulRedisClusterConnectionImpl) {
                    ((StatefulRedisClusterConnectionImpl) connection).dispatch(command);
                }

                if (connection instanceof StatefulRedisConnectionImpl) {
                    ((StatefulRedisConnectionImpl) connection).dispatch(command);
                }

                return command;
            });

        }

        if (LettuceStrings.isNotEmpty(connectionSettings.getClientName())) {
            sync = sync.thenApply(channelHandler -> {

                if (connection instanceof StatefulRedisClusterConnectionImpl) {
                    ((StatefulRedisClusterConnectionImpl) connection).setClientName(connectionSettings.getClientName());
                }

                if (connection instanceof StatefulRedisConnectionImpl) {
                    ((StatefulRedisConnectionImpl) connection).setClientName(connectionSettings.getClientName());
                }
                return channelHandler;
            });
        }

        return sync.thenApply(channelHandler -> (S) connection);
    }