Wrap a parallel flowable.
@param source The source
@param instrumentations The instrumentations
@param <T> The type
@return The wrapped subscriber
190554
static <T> ParallelFlowable<T> wrap(
            ParallelFlowable<T> source, Collection<ReactiveInstrumenter> instrumentations) {
        return new RxInstrumentedParallelFlowable<>(source, instrumentations);
    }
823028
@Override
    public BufferByteInput<T> source(final T source) {
        return (BufferByteInput<T>) super.source(source);
    }
479821
static Completable wrap(
            CompletableSource source, Collection<ReactiveInstrumenter> instrumentations) {
        if (source instanceof Callable) {
            return new RxInstrumentedCallableCompletable<>(source, instrumentations);
        }
        return new RxInstrumentedCompletable(source, instrumentations);
    }
543740
public static <T> Source<T> join(final Iterator<Source<T>> iterator) {
        return new AbstractSource<T>() {
            private Source<T> current = popSource();

            @Override
            public T computeNext() throws IOException {
                while (current != null) {
                    if (current.hasNext()) {
                        return current.next();
                    } else {
                        current.close();
                        current = popSource();
                    }
                }
                return endOfData();
            }

            @Override
            public void close() {
                while (current != null) {
                    current.close();
                    current = popSource();
                }
            }

            private Source<T> popSource() {
                if (iterator.hasNext()) {
                    return iterator.next();
                } else {
                    return null;
                }
            }
        };
    }
384372
<T> InternalProviderImpl<T> toProvider(Binding<T> binding) {
    if (binding == null) {
      throw new IllegalStateException("null binding are not allowed. Should not happen unless getBindingSet is overridden.");
    }
    ConfigurationHolder.configuration.checkIllegalBinding(binding, this);

    switch (binding.getMode()) {
      case SIMPLE:
        return createInternalProvider(this,
            binding.getKey(),
            false,
            binding.isCreatingInstancesInScope(),
            binding.isCreatingSingletonInScope(),
            false);
      case CLASS:
        return createInternalProvider(this,
            binding.getImplementationClass(),
            false,
            binding.isCreatingInstancesInScope(),
            binding.isCreatingSingletonInScope(),
            false);
      case INSTANCE:
        return new InternalProviderImpl<>(binding.getInstance());
      case PROVIDER_INSTANCE:
        // to ensure providers do not have to deal with concurrency, we wrap them in a thread safe provider
        // We do not need to pass the scope here because the provider won't use any scope to create the instance
        return new InternalProviderImpl<>(binding.getProviderInstance(), binding.isProvidingSingletonInScope());
      case PROVIDER_CLASS:
        return createInternalProvider(this,
            binding.getProviderClass(),
            true,
            binding.isCreatingInstancesInScope(),
            binding.isCreatingSingletonInScope(),
            binding.isProvidingSingletonInScope());
      //JACOCO:OFF
      default:
        throw new IllegalStateException(format("mode is not handled: %s. This should not happen.", binding.getMode()));
        //JACOCO:ON
    }
  }
93775
@Override
  public Dag<JobExecutionPlan> compileFlow(Spec spec) {
    Preconditions.checkNotNull(spec);
    Preconditions.checkArgument(spec instanceof FlowSpec, "MultiHopFlowCompiler only accepts FlowSpecs");

    long startTime = System.nanoTime();

    FlowSpec flowSpec = (FlowSpec) spec;
    String source = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, "");
    String destination =
        ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, "");
    log.info(String.format("Compiling flow for source: %s and destination: %s", source, destination));

    Dag<JobExecutionPlan> jobExecutionPlanDag;
    try {
      //Compute the path from source to destination.
      FlowGraphPath flowGraphPath = flowGraph.findPath(flowSpec);
      //Convert the path into a Dag of JobExecutionPlans.
      if (flowGraphPath != null) {
        jobExecutionPlanDag = flowGraphPath.asDag(this.config);
      } else {
        Instrumented.markMeter(flowCompilationFailedMeter);
        log.info(String.format("No path found from source: %s and destination: %s", source, destination));
        return new JobExecutionPlanDagFactory().createDag(new ArrayList<>());
      }
    } catch (PathFinder.PathFinderException | SpecNotFoundException | JobTemplate.TemplateException | URISyntaxException | ReflectiveOperationException e) {
      Instrumented.markMeter(flowCompilationFailedMeter);
      log.error(String
              .format("Exception encountered while compiling flow for source: %s and destination: %s", source, destination),
          e);
      return null;
    }
    Instrumented.markMeter(flowCompilationSuccessFulMeter);
    Instrumented.updateTimer(flowCompilationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
    return jobExecutionPlanDag;
  }
202101
public static <A> IO<A> io(Supplier<A> supplier) {
        return new IO<A>() {
            @Override
            public A unsafePerformIO() {
                return supplier.get();
            }

            @Override
            public CompletableFuture<A> unsafePerformAsyncIO(Executor executor) {
                return supplyAsync(supplier, executor);
            }
        };
    }
1352467
public static InjectionProvider<?>[] providersForInstancesSuppliers(final InjectionProviderInstancesSupplier... suppliers) {
        final InjectionProviderInstancesSupplier supplier = mergeSuppliers(suppliers);
        return supplier.get().toArray(new InjectionProvider<?>[supplier.get().size()]);
    }
909885
@SuppressWarnings("unchecked")
  @Override
  public void sink(FlowProcess<? extends Configuration> flowProcess, SinkCall<Corc, OutputCollector> sinkCall)
      throws IOException {
    Corc corc = sinkCall.getContext();
    TupleEntry tupleEntry = sinkCall.getOutgoingEntry();
    for (Comparable<?> fieldName : tupleEntry.getFields()) {
      corc.set(fieldName.toString(), tupleEntry.getObject(fieldName));
    }
    sinkCall.getOutput().collect(null, corc);
  }
220040
static <T> SingleObserver<T> wrap(SingleObserver<T> downstream,
                                      List<RunnableInstrumenter> instrumentations) {
        return new RxInstrumentedSingleObserver<>(downstream, instrumentations);
    }
489441
private <X> DataSource<X> fromParallelCollection(SplittableIterator<X> iterator, TypeInformation<X> type, String callLocationName) {
		return new DataSource<>(this, new ParallelIteratorInputFormat<>(iterator), type, callLocationName);
	}
963909
Dispatcher subscribeImpl(String subject, String queueName) {
        if (!this.running.get()) {
            throw new IllegalStateException("Dispatcher is closed");
        }
        
        if (this.isDraining()) {
            throw new IllegalStateException("Dispatcher is draining");
        }

        NatsSubscription sub = subscriptions.get(subject);

        if (sub == null) {
            sub = connection.createSubscription(subject, queueName, this);
            NatsSubscription actual = subscriptions.putIfAbsent(subject, sub);
            if (actual != null) {
                this.connection.unsubscribe(sub, -1); // Could happen on very bad timing
            }
        }

        return this;
    }
664348
public static <T> Flowable<T> toFlowable(EventPublisher<T> eventPublisher) {
        PublishProcessor<T> publishProcessor = PublishProcessor.create();
        FlowableProcessor<T> flowableProcessor = publishProcessor.toSerialized();
        eventPublisher.onEvent(flowableProcessor::onNext);
        return flowableProcessor;
    }
175696
protected void registerInstrumentationService(BundleContext systemContext) {
        Instrumentation inst = config.getInstrumentation();
        if (inst != null) {
            // Register a wrapper so we can trace callers.
            inst = (Instrumentation) Proxy.newProxyInstance(TraceInstrumentation.class.getClassLoader(),
                                                            new Class[] { Instrumentation.class },
                                                            new TraceInstrumentation(inst));
            Hashtable<String, String> svcProps = new Hashtable<String, String>();
            systemContext.registerService(Instrumentation.class.getName(), inst, svcProps);
        }
    }
1238197
public static <T> FlowableTransformer<T,T> doOnSubscribeFx(Consumer<Subscription> subscribe) {
        return obs -> obs.doOnSubscribe((d -> runOnFx(d,subscribe)));
    }
1035612
public <R> R wrapClassLoader(Supplier<R> supplier) {
		try (TemporaryClassLoaderContext tmpCl = new TemporaryClassLoaderContext(classLoader)){
			return supplier.get();
		}
	}
162019
public void connectSource(int inputNumber, IntermediateResult source, JobEdge edge, int consumerNumber) {

		final DistributionPattern pattern = edge.getDistributionPattern();
		final IntermediateResultPartition[] sourcePartitions = source.getPartitions();

		ExecutionEdge[] edges;

		switch (pattern) {
			case POINTWISE:
				edges = connectPointwise(sourcePartitions, inputNumber);
				break;

			case ALL_TO_ALL:
				edges = connectAllToAll(sourcePartitions, inputNumber);
				break;

			default:
				throw new RuntimeException("Unrecognized distribution pattern.");

		}

		inputEdges[inputNumber] = edges;

		// add the consumers to the source
		// for now (until the receiver initiated handshake is in place), we need to register the
		// edges as the execution graph
		for (ExecutionEdge ee : edges) {
			ee.getSource().addConsumer(ee, consumerNumber);
		}
	}