akka.stream.javadsl.Source.fromPublisher()方法的使用及代码示例

x33g5p2x  于2022-01-30 转载在 其他  
字(10.6k)|赞(0)|评价(0)|浏览(100)

本文整理了Java中akka.stream.javadsl.Source.fromPublisher()方法的一些代码示例，展示了Source.fromPublisher()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Source.fromPublisher()方法的具体详情如下：
包路径：akka.stream.javadsl.Source
类名称：Source
方法名：fromPublisher

Source.fromPublisher介绍

暂无

代码示例

代码示例来源：origin: eclipse/ditto

@Override
public Source<NotUsed, NotUsed> updateLastSuccessfulStreamEnd(final Instant timestamp) {
  final Date mongoStorableDate = Date.from(timestamp);
  final Document toStore = new Document()
      .append(FIELD_TIMESTAMP, mongoStorableDate);
  return Source.fromPublisher(lastSuccessfulSearchSyncCollection.insertOne(toStore))
      .map(success -> {
        LOGGER.debug("Successfully inserted timestamp for search synchronization: <{}>.", timestamp);
        return NotUsed.getInstance();
      });
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-utils-persistence

@Override
public Source<NotUsed, NotUsed> updateLastSuccessfulStreamEnd(final Instant timestamp) {
  final Date mongoStorableDate = Date.from(timestamp);
  final Document toStore = new Document()
      .append(FIELD_TIMESTAMP, mongoStorableDate);
  return Source.fromPublisher(lastSuccessfulSearchSyncCollection.insertOne(toStore))
      .map(success -> {
        LOGGER.debug("Successfully inserted timestamp for search synchronization: <{}>.", timestamp);
        return NotUsed.getInstance();
      });
}

代码示例来源：origin: eclipse/ditto

private Source<Optional<Instant>, NotUsed> retrieveLastSuccessfulStreamEndAsync() {
  return Source.fromPublisher(lastSuccessfulSearchSyncCollection.find())
      .limit(1)
      .flatMapConcat(doc -> {
        final Date date = doc.getDate(FIELD_TIMESTAMP);
        final Instant timestamp = date.toInstant();
        LOGGER.debug("Returning last timestamp of search synchronization: <{}>.", timestamp);
        return Source.single(Optional.of(timestamp));
      })
      .orElse(Source.single(Optional.empty()));
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-utils-persistence

private Source<Optional<Instant>, NotUsed> retrieveLastSuccessfulStreamEndAsync() {
  return Source.fromPublisher(lastSuccessfulSearchSyncCollection.find())
      .limit(1)
      .flatMapConcat(doc -> {
        final Date date = doc.getDate(FIELD_TIMESTAMP);
        final Instant timestamp = date.toInstant();
        LOGGER.debug("Returning last timestamp of search synchronization: <{}>.", timestamp);
        return Source.single(Optional.of(timestamp));
      })
      .orElse(Source.single(Optional.empty()));
}

代码示例来源：origin: com.typesafe.play/play-streams_2.12

private BiFunction<Accumulator<E, A>, Throwable, Void> handler(final Subscription sub) {
  return (acc, error) -> {
    if (acc != null) {
      Source.fromPublisher(publisher(sub)).runWith(acc.toSink().mapMaterializedValue(this::completeResultWith), materializer);
    } else {
      // On error
      sub.cancel();
      result.completeExceptionally(error);
    }
    return null;
  };
}

代码示例来源：origin: com.typesafe.play/play-streams_2.11

private BiFunction<Accumulator<E, A>, Throwable, Void> handler(final Subscription sub) {
  return (acc, error) -> {
    if (acc != null) {
      Source.fromPublisher(publisher(sub)).runWith(acc.toSink().mapMaterializedValue(this::completeResultWith), materializer);
    } else {
      // On error
      sub.cancel();
      result.completeExceptionally(error);
    }
    return null;
  };
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-thingsearch-persistence

@Override
public Source<Document, NotUsed> execute(final MongoCollection<Document> collection, final Duration maxTime) {
  checkNotNull(collection, "collection to be aggregated");
  final Publisher<Document> publisher = collection.aggregate(aggregationPipeline)
      .maxTime(maxTime.getSeconds(), TimeUnit.SECONDS)
      .allowDiskUse(true)
      // query is faster with cursor disabled
      .useCursor(false);
  return Source.<Document>fromPublisher(publisher);
}

代码示例来源：origin: eclipse/ditto

@Override
public Source<Document, NotUsed> execute(final MongoCollection<Document> collection, final Duration maxTime) {
  checkNotNull(collection, "collection to be aggregated");
  final Publisher<Document> publisher = collection.aggregate(aggregationPipeline)
      .maxTime(maxTime.getSeconds(), TimeUnit.SECONDS)
      .allowDiskUse(true)
      // query is faster with cursor disabled
      .useCursor(false);
  return Source.<Document>fromPublisher(publisher);
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-utils-persistence

/**
 * Drops the specified index. Does <strong>not</strong> throw an exception if the index does not exist.
 *
 * @param collectionName the name of the collection containing the index.
 * @param indexName the name of the index.
 * @return a source which emits {@link Success}.
 */
public Source<Success, NotUsed> dropIndex(final String collectionName, final String indexName) {
  return Source.fromPublisher(getCollection(collectionName).dropIndex(indexName))
      .recoverWith(buildDropIndexRecovery(indexName));
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-thingsearch-persistence

private Source<Boolean, NotUsed> delete(final String thingId, final Bson filter, final Bson document) {
  return Source.fromPublisher(collection.updateOne(filter, document))
      .flatMapConcat(deleteResult -> {
        if (deleteResult.getMatchedCount() <= 0) {
          return Source.single(Boolean.FALSE);
        }
        final PolicyUpdate deletePolicyEntries = PolicyUpdateFactory.createDeleteThingUpdate(thingId);
        final Bson policyIndexRemoveFilter = deletePolicyEntries.getPolicyIndexRemoveFilter();
        return Source.fromPublisher(policiesCollection.deleteMany(policyIndexRemoveFilter))
            .map(r -> true);
      });
}

代码示例来源：origin: eclipse/ditto

private Source<Boolean, NotUsed> delete(final String thingId, final Bson filter, final Bson document) {
  return Source.fromPublisher(collection.updateOne(filter, document))
      .flatMapConcat(deleteResult -> {
        if (deleteResult.getMatchedCount() <= 0) {
          return Source.single(Boolean.FALSE);
        }
        final PolicyUpdate deletePolicyEntries = PolicyUpdateFactory.createDeleteThingUpdate(thingId);
        final Bson policyIndexRemoveFilter = deletePolicyEntries.getPolicyIndexRemoveFilter();
        return Source.fromPublisher(policiesCollection.deleteMany(policyIndexRemoveFilter))
            .map(r -> true);
      });
}

代码示例来源：origin: eclipse/ditto

/**
 * Drops the specified index. Does <strong>not</strong> throw an exception if the index does not exist.
 *
 * @param collectionName the name of the collection containing the index.
 * @param indexName the name of the index.
 * @return a source which emits {@link Success}.
 */
public Source<Success, NotUsed> dropIndex(final String collectionName, final String indexName) {
  return Source.fromPublisher(getCollection(collectionName).dropIndex(indexName))
      .recoverWith(buildDropIndexRecovery(indexName));
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-thingsearch-persistence

@Override
public final Source<Set<String>, NotUsed> getThingIdsForPolicy(final String policyId) {
  log.debug("Retrieving Thing ids for policy: <{}>", policyId);
  final Bson filter = eq(FIELD_POLICY_ID, policyId);
  return Source.fromPublisher(collection.find(filter)
      .projection(new BsonDocument(FIELD_ID, new BsonInt32(1))))
      .map(doc -> doc.getString(FIELD_ID))
      .fold(new HashSet<>(), (set, id) -> {
        set.add(id);
        return set;
      });
}

代码示例来源：origin: eclipse/ditto

@Override
public final Source<Set<String>, NotUsed> getThingIdsForPolicy(final String policyId) {
  log.debug("Retrieving Thing ids for policy: <{}>", policyId);
  final Bson filter = eq(FIELD_POLICY_ID, policyId);
  return Source.fromPublisher(collection.find(filter)
      .projection(new BsonDocument(FIELD_ID, new BsonInt32(1))))
      .map(doc -> doc.getString(FIELD_ID))
      .fold(new HashSet<>(), (set, id) -> {
        set.add(id);
        return set;
      });
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-thingsearch-persistence

/**
 * {@inheritDoc}
 */
@Override
public final Source<Boolean, NotUsed> updatePolicy(final Thing thing, final Enforcer policyEnforcer) {
  log.debug("Updating policy for Thing: <{}>", thing);
  final PolicyUpdate policyUpdate = PolicyUpdateFactory.createPolicyIndexUpdate(thing, policyEnforcer);
  return Source.fromPublisher(updatePolicy(thing, policyUpdate))
      .flatMapConcat(mapPolicyUpdateResult(policyUpdate))
      .recoverWithRetries(1, errorRecovery(getThingId(thing)));
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-thingsearch-persistence

/**
 * {@inheritDoc}
 */
@Override
public final Source<ThingMetadata, NotUsed> getThingMetadata(final String thingId) {
  log.debug("Retrieving Thing Metadata for Thing: <{}>", thingId);
  final Bson filter = eq(FIELD_ID, thingId);
  return Source.fromPublisher(collection.find(filter)
      .projection(Projections.include(FIELD_REVISION, FIELD_POLICY_ID, FIELD_POLICY_REVISION)))
      .map(mapThingMetadataToModel())
      .orElse(defaultThingMetadata());
}

代码示例来源：origin: eclipse/ditto

/**
 * {@inheritDoc}
 */
@Override
public final Source<ThingMetadata, NotUsed> getThingMetadata(final String thingId) {
  log.debug("Retrieving Thing Metadata for Thing: <{}>", thingId);
  final Bson filter = eq(FIELD_ID, thingId);
  return Source.fromPublisher(collection.find(filter)
      .projection(Projections.include(FIELD_REVISION, FIELD_POLICY_ID, FIELD_POLICY_REVISION)))
      .map(mapThingMetadataToModel())
      .orElse(defaultThingMetadata());
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-thingsearch-persistence

/**
 * {@inheritDoc}
 */
@Override
protected final Source<Boolean, NotUsed> save(final Thing thing, final long revision, final long policyRevision) {
  log.debug("Saving Thing with revision <{}> and policy revision <{}>: <{}>", revision, policyRevision, thing);
  final Bson filter =
      filterWithLowerThingRevisionOrLowerPolicyRevision(getThingId(thing), revision, policyRevision);
  final Document document = toUpdate(ThingDocumentMapper.toDocument(thing), revision, policyRevision);
  return Source.fromPublisher(collection.updateOne(filter, document, new UpdateOptions().upsert(true)))
      .map(updateResult -> updateResult.getMatchedCount() > 0 || null != updateResult.getUpsertedId());
}

代码示例来源：origin: eclipse/ditto

/**
 * {@inheritDoc}
 */
@Override
protected final Source<Boolean, NotUsed> save(final Thing thing, final long revision, final long policyRevision) {
  log.debug("Saving Thing with revision <{}> and policy revision <{}>: <{}>", revision, policyRevision, thing);
  final Bson filter =
      filterWithLowerThingRevisionOrLowerPolicyRevision(getThingId(thing), revision, policyRevision);
  final Document document = toUpdate(ThingDocumentMapper.toDocument(thing), revision, policyRevision);
  return Source.fromPublisher(collection.updateOne(filter, document, new UpdateOptions().upsert(true)))
      .map(updateResult -> updateResult.getMatchedCount() > 0 || null != updateResult.getUpsertedId());
}

代码示例来源：origin: org.eclipse.ditto/ditto-services-thingsearch-persistence

@Override
public Source<String, NotUsed> getOutdatedThingIds(final PolicyTag policyTag) {
  log.debug("Retrieving outdated Thing ids with policy tag: <{}>", policyTag);
  final String policyId = policyTag.getId();
  final Bson filter = and(eq(FIELD_POLICY_ID, policyId), lt(FIELD_POLICY_REVISION, policyTag.getRevision()));
  final Publisher<Document> publisher =
      collection.find(filter).projection(new BsonDocument(FIELD_ID, new BsonInt32(1)));
  return Source.fromPublisher(publisher)
      .map(doc -> doc.getString(FIELD_ID));
}

内容来源于网络，如有侵权，请联系作者删除！

source

相关文章

    查看更多
    微信公众号

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com