rx.Observable.count()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(8.3k)|赞(0)|评价(0)|浏览(189)

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

Observable.count介绍

[英]Returns an Observable that emits the count of the total number of items emitted by the source Observable.

Backpressure Support: This operator does not support backpressure because by intent it will receive all values and reduce them to a single onNext. Scheduler: count does not operate by default on a particular Scheduler.
[中]返回一个Observable，该Observable发出源Observable发出的项目总数的计数。
背压支持：此运算符不支持背压，因为它会接收所有值，并将它们减少到单个onNext。计划程序：默认情况下，计数不会在特定计划程序上运行。

代码示例

代码示例来源：origin: apache/usergrid

.doOnNext( RX_LOG ).take( 1 ).count()
.doOnNext( count -> {

代码示例来源：origin: apache/usergrid

@Test
@Category(ExperimentalTest.class )
public void testConnectableObserver() throws InterruptedException {
  final int count = 10;
  final CountDownLatch latch = new CountDownLatch( count );
  final ConnectableObservable<Integer> connectedObservable = Observable.range( 0, count ).publish();
  //connect to our latch, which should run on it's own subscription
  //start our latch running
  connectedObservable.doOnNext( integer -> latch.countDown() ).subscribeOn( Schedulers.io() ).subscribe();
  final Observable<Integer> countObservable = connectedObservable.subscribeOn( Schedulers.io() ).count();
  //start the sequence
  connectedObservable.connect();
  final boolean completed = latch.await( 5, TimeUnit.SECONDS );
  assertTrue( "publish1 behaves as expected", completed );
  final int returnedCount = countObservable.toBlocking().last();
  assertEquals( "Counts the same", count, returnedCount );
}

代码示例来源：origin: apache/usergrid

} ).count().defaultIfEmpty( 0 ).toBlocking().last();

代码示例来源：origin: apache/usergrid

@Test
public void testGetEdgesToTarget() {
  final GraphManager gm = emf.createEdgeManager( scope );
  Id sourceId1 = new SimpleId( "source1" );
  Id sourceId2 = new SimpleId( "source2" );
  Id targetId1 = new SimpleId( "target" );
  Edge testTargetEdge = createEdge( sourceId1, "test", targetId1, System.currentTimeMillis() );
  gm.writeEdge( testTargetEdge ).toBlocking().singleOrDefault( null );
  Edge testTarget2Edge = createEdge( sourceId2, "edgeType1", targetId1, System.currentTimeMillis() );
  gm.writeEdge( testTarget2Edge ).toBlocking().singleOrDefault( null );
  Edge test2TargetEdge = createEdge( sourceId1, "edgeType1", targetId1, System.currentTimeMillis() );
  gm.writeEdge( test2TargetEdge ).toBlocking().singleOrDefault( null );
  Edge test3TargetEdge = createEdge( sourceId1, "edgeType2", targetId1, System.currentTimeMillis() );
  gm.writeEdge( test3TargetEdge ).toBlocking().singleOrDefault( null );
  int count = gm.getEdgeTypesToTarget( new SimpleSearchEdgeType(targetId1, null, null) )
      .count().toBlocking().last();
  assertEquals( 3, count );
  count = gm.getEdgeTypesToTarget( new SimpleSearchEdgeType(targetId1, "edgeType", null) )
      .count().toBlocking().last();
  assertEquals( 2, count );
}

代码示例来源：origin: apache/usergrid

/**
 * Simple test case that tests a single edge and removing the node.  The other target node should be removed as well
 * since it has no other targets
 */
@Test
public void testNoDeletionMarked() {
  GraphManager em = emf.createEdgeManager( scope );
  Edge edge = createEdge( "source", "test", "target" );
  //write the edge
  Edge last = em.writeEdge( edge ).toBlocking().last();
  assertEquals( edge, last );
  Id sourceNode = edge.getSourceNode();
  UUID eventTime = UUIDGenerator.newTimeUUID();
  int count = deleteListener.receive( scope, sourceNode, eventTime ).count().toBlocking().last();
  assertEquals( "Mark was not set, no delete should be executed", 0, count );
}

代码示例来源：origin: apache/usergrid

Observable<Id> ids =
  this.app.getApplicationService().deleteAllEntities(appScope, 5);
int count = ids.count().toBlocking().last();
Assert.assertEquals(count, 5);
ids =
  this.app.getApplicationService().deleteAllEntities(appScope, 5);
count = ids.count().toBlocking().last();
Assert.assertEquals(count, 5);
this.app.waitForQueueDrainAndRefreshIndex();

代码示例来源：origin: apache/usergrid

@Test
public void testSingleConnection() {
  final ApplicationScope applicationScope = new ApplicationScopeImpl( new SimpleId( "application" ) );
  final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope );
  //now write a single connection
  final Id source = new SimpleId( "source" );
  //add to a collection
  final String collectionName = "testCollection";
  final Edge collectionEdge = CpNamingUtils.createCollectionEdge( applicationScope.getApplication(), collectionName, source );
  final Edge writtenCollection = gm.writeEdge( collectionEdge ).toBlocking().last();
  assertNotNull("Collection edge written", writtenCollection);
  final Id target = new SimpleId( "target" );
  final String connectionType = "testConnection";
  final Edge connectionEdge = CpNamingUtils.createConnectionEdge( source, connectionType, target );
  final Edge writtenConnection = gm.writeEdge( connectionEdge ).toBlocking().last();
  //now run the cleanup
  final int count =
    connectionService.deDupeConnections( Observable.just( applicationScope ) ).count().toBlocking().last();
  assertEquals( "No edges deleted", 0, count );
  //now ensure we can read the edge.
  final SearchByEdge simpleSearchByEdge =
    new SimpleSearchByEdge( source, connectionEdge.getType(), target, Long.MAX_VALUE,
      SearchByEdgeType.Order.DESCENDING, Optional.absent() );
  final List<MarkedEdge> edges = gm.loadEdgeVersions( simpleSearchByEdge ).toList().toBlocking().last();
  assertEquals( 1, edges.size() );
  assertEquals( writtenConnection, edges.get( 0 ) );
}

代码示例来源：origin: apache/usergrid

int appCount =  Observable.from( applicationInfoResults.getEntities() ).filter(
  entity -> !entity.getName().startsWith( "org." )  ).doOnNext(
  entity -> logger.info("counting entity {}", entity) ).count().toBlocking().last();
assertEquals( appIds.size() ,appCount );

代码示例来源：origin: apache/usergrid

int count = deleteListener.receive( scope, toDelete, UUIDGenerator.newTimeUUID() ).count().toBlocking().last();

代码示例来源：origin: apache/usergrid

int count = deleteListener.receive( scope, targetNode, UUIDGenerator.newTimeUUID() ).count().toBlocking().last();

代码示例来源：origin: apache/usergrid

int count = deleteListener.receive( scope, sourceNode, deleteEventTimestamp ).count().toBlocking().last();

代码示例来源：origin: apache/usergrid

.flatMap(mesage ->indexProducer.put(mesage)).count().toBlocking().last();

代码示例来源：origin: davidmoten/rxjava-jdbc

public Observable<Integer> count() {
  return get(Util.toOne()).count();
}

代码示例来源：origin: davidmoten/rxjava-jdbc

public Observable<Integer> count() {
  return get(Util.toOne()).count();
}

代码示例来源：origin: apache/usergrid

.flatMap(mesage -> indexProducer.put(mesage)).count().toBlocking().last();

代码示例来源：origin: apache/usergrid

.flatMap(mesage -> indexProducer.put(mesage)).count().toBlocking().last();

代码示例来源：origin: au.gov.amsa.risky/formats

public static void main(String[] args) throws InterruptedException {
  String output = System.getProperty("output", "target/output");
  // output = "/media/an/binary-fixes-2012/temp";
  long sampleSeconds = Long.parseLong(System.getProperty("sampleSeconds", "0"));
  BinaryFixes
      .sortBinaryFixFilesByTime(new File(output), sampleSeconds, Schedulers.immediate())
      .count().toBlocking().single();
}

代码示例来源：origin: au.gov.amsa.risky/formats

public static void main(String[] args) {
  final File input = new File(System.getProperty("input"));
  final File output = new File(System.getProperty("output"));
  Pattern pattern = Pattern.compile(System.getProperty("pattern"));
  Action2<List<HasFix>, File> fixesWriter = (fixes, file) -> {
    BinaryFixesWriter.writeFixes(fixes, file, false, true, BinaryFixesFormat.WITHOUT_MMSI);
  };
  Func1<String, String> renamer = name -> name + ".zip";
  Formats.transform(input, output, pattern, Transformers.<HasFix> identity(), fixesWriter,
      renamer).count().toBlocking().single();
}

代码示例来源：origin: au.gov.amsa.risky/formats

public static void sort(File output) {
  // now sort the data in each output file by time and rewrite
  List<File> files = Files.find(output, Pattern.compile(".*\\.fix"));
  Observable.from(files)
      //
      .buffer(Math.max(1, files.size() / Runtime.getRuntime().availableProcessors()))
      //
      .flatMap(list -> Observable.from(list)
          //
          .doOnNext(file -> sortFixFile(file)).subscribeOn(Schedulers.computation()))
      .count().toBlocking().single();
}

代码示例来源：origin: nurkiewicz/rxjava-book-examples

@Test
public void sample_216() throws Exception {
  Observable<KeyEvent> keyEvents = empty();
  Observable<Observable<KeyEvent>> windows = keyEvents.window(1, SECONDS);
  Observable<Integer> eventPerSecond = windows
      .flatMap(eventsInSecond -> eventsInSecond.count());
}

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

Observable

相关文章

    查看更多

    热门标签

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

    最新文章

    更多

    Observable类方法

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