Synchronously publish data to the Adapter specified by the provided Key, blocking the current thread

@param key for registered cylops-react async.Adapter
@param publisher Reactive Streams publisher  to push data onto this pipe
352405
public void publishTo(final K key, final Publisher<V> publisher) {
        registered.get(key).fromStream(Spouts.from(publisher));
    }
1087253
function Publisher (ossConfig, cacheOptions) {
  this.config = ossConfig;
  this.client = new OSS(ossConfig.connect);
  this._newCache = {};
  var bucket = this.config.connect.bucket;

  if (!bucket) {
    throw new Error('Missing `connect.bucket` config value.');
  }

  // init Cache file
  this._cacheFile = cacheOptions && cacheOptions.cacheFileName
    ? cacheOptions.cacheFileName
    : '.oss-cache-' + bucket;

  // load cache
  if (ossConfig.setting.force) {
    try {
      fs.unlinkSync(this.getCacheFilename());
    } catch (err) {} finally {
      this._oldCache = {};
    }
  } else {
    try {
      this._oldCache = JSON.parse(fs.readFileSync(this.getCacheFilename(), 'utf8'));
    } catch (err) {
      this._oldCache = {};
    }
  }
}
303277
public static void addAsyncProperty(String key, String value, TransactionLogger transactionLogger) {

    FlowContextFactory.deserializeNativeFlowContext(transactionLogger.getFlowContextAsync(transactionLogger));
    transactionLogger.properties.put(key, value);

  }
314268
public static <T, R> Publisher<R> map(Publisher<T> publisher, Function<T, R> mapper) {
        return actual -> publisher.subscribe(new CompletionAwareSubscriber<T>() {
            @Override
            protected void doOnSubscribe(Subscription subscription) {
                actual.onSubscribe(subscription);
            }

            @Override
            protected void doOnNext(T message) {
                try {
                    R result = Objects.requireNonNull(mapper.apply(message),
                        "The mapper returned a null value.");
                    actual.onNext(result);
                } catch (Throwable e) {
                    onError(e);
                }

            }

            @Override
            protected void doOnError(Throwable t) {
                actual.onError(t);
            }

            @Override
            protected void doOnComplete() {
                actual.onComplete();
            }
        });
    }
129251
public void createBlob(String key, byte [] data, SettableBlobMeta meta) throws KeyAlreadyExistsException, IOException {
        AtomicOutputStream out = null;
        try {
            out = createBlob(key, meta);
            out.write(data);
            out.close();
            out = null;
        } finally {
            if (out != null) {
                out.cancel();
            }
        }
    }
1359585
public <E extends Event> void publish(E event) {
        if (!eventListenerMap.isEmpty()) {
            EventListener<E> eventListener = getEventListener(event);
            if (eventListener != null) {
                eventListener.on(event);
            }
        }
    }
1217281
public <T> SimpleReactStream<T> fromPublisher(final Publisher<? extends T> publisher) {
        Objects.requireNonNull(publisher);
        Publisher<T> narrowed = (Publisher<T>)publisher;
        return from(Spouts.from(narrowed));
    }
388074
public void register(K key, T connection) {
        connections.put(key, new Sync<>(key, connection));
    }
1176738
protected void publishWanEvent(String mapName, ReplicationEventObject event) {
        MapContainer mapContainer = mapServiceContext.getMapContainer(mapName);
        WanReplicationPublisher wanReplicationPublisher = mapContainer.getWanReplicationPublisher();
        if (isOwnedPartition(event.getKey())) {
            wanReplicationPublisher.publishReplicationEvent(SERVICE_NAME, event);
        } else {
            wanReplicationPublisher.publishReplicationEventBackup(SERVICE_NAME, event);
        }
    }
755054
public <T> FutureStream<T> fromPublisher(final Publisher<? extends T> publisher) {
        Objects.requireNonNull(publisher);
        Publisher<T> narrowed = (Publisher<T>)publisher;
        return fromStream(Spouts.from(narrowed));
    }
132577
@Override
    public void registerEarlyRead(InterChannelCallback cb) {
        // LI4335.2
        if (null == cb) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "registerEarlyRead: Invalid null callback as input.");
            }
            throw new IllegalArgumentException("Callback is null");
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "App channel requesting early response read on " + getVC());
        }
        this.bEarlyReads = true;
        startEarlyRead(cb, true);
    }
640742
void publishTaskState(String taskName, Map stateSnapshot, ScheduledTaskStatisticsImpl statsSnapshot,
                          ScheduledTaskResult result) {
        if (logger.isFinestEnabled()) {
            log(FINEST, "Publishing state, to replicas. State: " + stateSnapshot);
        }

        Operation op = new SyncStateOperation(getName(), taskName, stateSnapshot, statsSnapshot, result);
        createInvocationBuilder(op)
                .invoke()
                .join();
    }
243937
private void publish(Cache<K, V> cache, EventType eventType,
      K key, @Nullable V oldValue, @Nullable V newValue, boolean quiet) {
    if (dispatchQueues.isEmpty()) {
      return;
    }

    JCacheEntryEvent<K, V> event = null;
    for (Registration<K, V> registration : dispatchQueues.keySet()) {
      if (!registration.getCacheEntryListener().isCompatible(eventType)) {
        continue;
      }
      if (event == null) {
        event = new JCacheEntryEvent<>(cache, eventType, key, oldValue, newValue);
      }
      if (!registration.getCacheEntryFilter().evaluate(event)) {
        continue;
      }

      JCacheEntryEvent<K, V> e = event;
      CompletableFuture<Void> future =
          dispatchQueues.computeIfPresent(registration, (k, queue) -> {
            Runnable action = () -> registration.getCacheEntryListener().dispatch(e);
            return queue.thenRunAsync(action, executor);
          });
      if ((future != null) && registration.isSynchronous() && !quiet) {
        pending.get().add(future);
      }
    }
  }
701891
public final <S, T> void registerConverter(ConverterKey<S,T> key, Converter<S, T> converter) {	

        if (key.getInputClass() == null) {
            throw new IllegalArgumentException("Input Class must not be null");
        }
        if (key.getOutputClass() == null) {
            throw new IllegalArgumentException("Output Class must not be null");
        }
        if (converter == null) {
            throw new IllegalArgumentException("Converter must not be null");
        }

        if (key.getQualifierAnnotation() == null) {
            throw new IllegalArgumentException("Qualifier must not be null");
        }

        Converter<?,?> old = registeredConverters.putIfAbsent(key, converter);
        if (old != null && (!isSameConverter(old, converter))) {
        	throw new IllegalStateException("Converter already registered for key: " + key);
        }
    }
1131955
@Override
	public void publish(LogRecord record) {
		if (!isLoggable(record)) {
			return;
		}
		synchronized (messages) {
			messages.addLast(record);
			messages.notifyAll();
		}
	}
765459
@Override
  protected void publishData(WorkUnitState state, int branchId, boolean publishSingleTaskData,
      Set<Path> writerOutputPathsMoved) throws IOException {
    Path publisherOutputDir = getPublisherOutputDir(state, branchId);
    if (!this.publisherFileSystemByBranches.get(branchId).exists(publisherOutputDir)) {
      WriterUtils.mkdirsWithRecursivePermissionWithRetry(this.publisherFileSystemByBranches.get(branchId),
          publisherOutputDir, this.permissions.get(branchId), this.retrierConfig);
    }
    super.publishData(state, branchId, publishSingleTaskData, writerOutputPathsMoved);
  }
57801
public <A> void publishEvent(final EventTranslatorOneArg<T, A> eventTranslator, final A arg)
    {
        ringBuffer.publishEvent(eventTranslator, arg);
    }