@Test @SuppressWarnings("unchecked") public void subscribeToPersistentSubnets_InvokeApi() { final int subnetId = 1; final UInt64 slot = UInt64.ONE; final SubnetSubscription subnetSubscription = new SubnetSubscription(subnetId, slot); final tech.pegasys.teku.api.schema.SubnetSubscription schemaSubnetSubscription = new tech.pegasys.teku.api.schema.SubnetSubscription(subnetId, slot); ArgumentCaptor<Set<tech.pegasys.teku.api.schema.SubnetSubscription>> argumentCaptor = ArgumentCaptor.forClass(Set.class); apiHandler.subscribeToPersistentSubnets(Set.of(subnetSubscription)); asyncRunner.executeQueuedActions(); verify(apiClient).subscribeToPersistentSubnets(argumentCaptor.capture()); final Set<tech.pegasys.teku.api.schema.SubnetSubscription> request = argumentCaptor.getValue(); assertThat(request).hasSize(1); assertThat(request.stream().findFirst().get()) .usingRecursiveComparison() .isEqualTo(schemaSubnetSubscription); }
@Override public void subscribeToPersistentSubnets(final Set<SubnetSubscription> subnetSubscriptions) { final Set<tech.pegasys.teku.api.schema.SubnetSubscription> schemaSubscriptions = subnetSubscriptions.stream() .map( s -> new tech.pegasys.teku.api.schema.SubnetSubscription( s.getSubnetId(), s.getUnsubscriptionSlot())) .collect(Collectors.toSet()); asyncRunner .runAsync(() -> apiClient.subscribeToPersistentSubnets(schemaSubscriptions)) .finish(error -> LOG.error("Failed to subscribe to persistent subnets", error)); }
RemoteValidatorApiHandler implements ValidatorApiChannel { @Override public void subscribeToPersistentSubnets(final Set<SubnetSubscription> subnetSubscriptions) { final Set<tech.pegasys.teku.api.schema.SubnetSubscription> schemaSubscriptions = subnetSubscriptions.stream() .map( s -> new tech.pegasys.teku.api.schema.SubnetSubscription( s.getSubnetId(), s.getUnsubscriptionSlot())) .collect(Collectors.toSet()); asyncRunner .runAsync(() -> apiClient.subscribeToPersistentSubnets(schemaSubscriptions)) .finish(error -> LOG.error("Failed to subscribe to persistent subnets", error)); } }
RemoteValidatorApiHandler implements ValidatorApiChannel { @Override public void subscribeToPersistentSubnets(final Set<SubnetSubscription> subnetSubscriptions) { final Set<tech.pegasys.teku.api.schema.SubnetSubscription> schemaSubscriptions = subnetSubscriptions.stream() .map( s -> new tech.pegasys.teku.api.schema.SubnetSubscription( s.getSubnetId(), s.getUnsubscriptionSlot())) .collect(Collectors.toSet()); asyncRunner .runAsync(() -> apiClient.subscribeToPersistentSubnets(schemaSubscriptions)) .finish(error -> LOG.error("Failed to subscribe to persistent subnets", error)); } RemoteValidatorApiHandler(final ServiceConfig config, final AsyncRunner asyncRunner); @VisibleForTesting  RemoteValidatorApiHandler(
      final ValidatorRestApiClient apiClient, final AsyncRunner asyncRunner); }
RemoteValidatorApiHandler implements ValidatorApiChannel { @Override public void subscribeToPersistentSubnets(final Set<SubnetSubscription> subnetSubscriptions) { final Set<tech.pegasys.teku.api.schema.SubnetSubscription> schemaSubscriptions = subnetSubscriptions.stream() .map( s -> new tech.pegasys.teku.api.schema.SubnetSubscription( s.getSubnetId(), s.getUnsubscriptionSlot())) .collect(Collectors.toSet()); asyncRunner .runAsync(() -> apiClient.subscribeToPersistentSubnets(schemaSubscriptions)) .finish(error -> LOG.error("Failed to subscribe to persistent subnets", error)); } RemoteValidatorApiHandler(final ServiceConfig config, final AsyncRunner asyncRunner); @VisibleForTesting  RemoteValidatorApiHandler(
      final ValidatorRestApiClient apiClient, final AsyncRunner asyncRunner); @Override SafeFuture<Optional<ForkInfo>> getForkInfo(); @Override SafeFuture<Optional<UInt64>> getGenesisTime(); @Override SafeFuture<Optional<List<ValidatorDuties>>> getDuties(
      final UInt64 epoch, final Collection<BLSPublicKey> publicKeys); @Override SafeFuture<Optional<List<AttesterDuties>>> getAttestationDuties(
      final UInt64 epoch, final Collection<Integer> validatorIndexes); @Override SafeFuture<Optional<List<ProposerDuties>>> getProposerDuties(final UInt64 epoch); @Override SafeFuture<Optional<Attestation>> createUnsignedAttestation(
      final UInt64 slot, final int committeeIndex); @Override void sendSignedAttestation(final Attestation attestation); @Override void sendSignedAttestation(
      final Attestation attestation, final Optional<Integer> validatorIndex); @Override SafeFuture<Optional<BeaconBlock>> createUnsignedBlock(
      final UInt64 slot, final BLSSignature randaoReveal, final Optional<Bytes32> graffiti); @Override void sendSignedBlock(final SignedBeaconBlock block); @Override SafeFuture<Optional<Attestation>> createAggregate(final Bytes32 attestationHashTreeRoot); @Override void sendAggregateAndProof(final SignedAggregateAndProof aggregateAndProof); @Override void subscribeToBeaconCommitteeForAggregation(
      final int committeeIndex, final UInt64 aggregationSlot); @Override void subscribeToPersistentSubnets(final Set<SubnetSubscription> subnetSubscriptions); }
RemoteValidatorApiHandler implements ValidatorApiChannel { @Override public void subscribeToPersistentSubnets(final Set<SubnetSubscription> subnetSubscriptions) { final Set<tech.pegasys.teku.api.schema.SubnetSubscription> schemaSubscriptions = subnetSubscriptions.stream() .map( s -> new tech.pegasys.teku.api.schema.SubnetSubscription( s.getSubnetId(), s.getUnsubscriptionSlot())) .collect(Collectors.toSet()); asyncRunner .runAsync(() -> apiClient.subscribeToPersistentSubnets(schemaSubscriptions)) .finish(error -> LOG.error("Failed to subscribe to persistent subnets", error)); } RemoteValidatorApiHandler(final ServiceConfig config, final AsyncRunner asyncRunner); @VisibleForTesting  RemoteValidatorApiHandler(
      final ValidatorRestApiClient apiClient, final AsyncRunner asyncRunner); @Override SafeFuture<Optional<ForkInfo>> getForkInfo(); @Override SafeFuture<Optional<UInt64>> getGenesisTime(); @Override SafeFuture<Optional<List<ValidatorDuties>>> getDuties(
      final UInt64 epoch, final Collection<BLSPublicKey> publicKeys); @Override SafeFuture<Optional<List<AttesterDuties>>> getAttestationDuties(
      final UInt64 epoch, final Collection<Integer> validatorIndexes); @Override SafeFuture<Optional<List<ProposerDuties>>> getProposerDuties(final UInt64 epoch); @Override SafeFuture<Optional<Attestation>> createUnsignedAttestation(
      final UInt64 slot, final int committeeIndex); @Override void sendSignedAttestation(final Attestation attestation); @Override void sendSignedAttestation(
      final Attestation attestation, final Optional<Integer> validatorIndex); @Override SafeFuture<Optional<BeaconBlock>> createUnsignedBlock(
      final UInt64 slot, final BLSSignature randaoReveal, final Optional<Bytes32> graffiti); @Override void sendSignedBlock(final SignedBeaconBlock block); @Override SafeFuture<Optional<Attestation>> createAggregate(final Bytes32 attestationHashTreeRoot); @Override void sendAggregateAndProof(final SignedAggregateAndProof aggregateAndProof); @Override void subscribeToBeaconCommitteeForAggregation(
      final int committeeIndex, final UInt64 aggregationSlot); @Override void subscribeToPersistentSubnets(final Set<SubnetSubscription> subnetSubscriptions);  }