FileBasedRegistrationService extends AbstractRegistrationService implements DeviceManagementService, RegistrationService, Lifecycle { @Override public Future<Void> start() { final Promise<Void> result = Promise.promise(); if (running.compareAndSet(false, true)) { if (!getConfig().isModificationEnabled()) { LOG.info("modification of registered devices has been disabled"); } if (getConfig().getFilename() == null) { LOG.debug("device identity filename is not set, no identity information will be loaded"); result.complete(); } else { checkFileExists(getConfig().isSaveToFile()) .compose(ok -> loadRegistrationData()) .onSuccess(ok -> { if (getConfig().isSaveToFile()) { LOG.info("saving device identities to file every 3 seconds"); vertx.setPeriodic(3000, tid -> saveToFile()); } else { LOG.info("persistence is disabled, will not save device identities to file"); } }) .onFailure(t -> { LOG.error("failed to start up service", t); running.set(false); }) .onComplete(result); } } else { result.complete(); } return result.future(); } @Autowired  FileBasedRegistrationService(final Vertx vertx); @Autowired void setConfig(final FileBasedRegistrationConfigProperties config); FileBasedRegistrationConfigProperties getConfig(); @Override Future<Void> start(); @Override Future<Void> stop(); @Override Future<OperationResult<Device>> readDevice(final String tenantId, final String deviceId, final Span span); @Override Future<Result<Void>> deleteDevice(final String tenantId, final String deviceId,
            final Optional<String> resourceVersion,
            final Span span); @Override Future<OperationResult<Id>> createDevice(
            final String tenantId, 
            final Optional<String> deviceId,
            final Device device,
            final Span span); OperationResult<Id> processCreateDevice(final String tenantId, final Optional<String> deviceId,
            final Device device, final Span span); @Override Future<OperationResult<Id>> updateDevice(final String tenantId, final String deviceId, final Device device,
            final Optional<String> resourceVersion, final Span span); @Override Future<OperationResult<SearchDevicesResult>> searchDevices(
            final String tenantId,
            final int pageSize,
            final int pageOffset,
            final List<Filter> filters,
            final List<Sort> sortOptions,
            final Span span); void clear(); @Override String toString();  }
@SuppressWarnings({ "unchecked" }) @Test public void testDoStartCreatesFile(final VertxTestContext ctx) { registrationConfig.setSaveToFile(true); when(fileSystem.existsBlocking(registrationConfig.getFilename())).thenReturn(Boolean.FALSE); doAnswer(invocation -> { final Handler<AsyncResult<Void>> handler = invocation.getArgument(1); handler.handle(Future.succeededFuture()); return null; }).when(fileSystem).createFile(eq(registrationConfig.getFilename()), any(Handler.class)); doAnswer(invocation -> { final Handler<AsyncResult<Buffer>> handler = invocation.getArgument(1); handler.handle(Future.failedFuture("malformed file")); return null; }).when(fileSystem).readFile(eq(registrationConfig.getFilename()), any(Handler.class)); final Promise<Void> startupTracker = Promise.promise(); startupTracker.future().onComplete(ctx.succeeding(started -> ctx.verify(() -> { verify(fileSystem).createFile(eq(registrationConfig.getFilename()), any(Handler.class)); ctx.completeNow(); }))); registrationService.start().onComplete(startupTracker); }
@SuppressWarnings({ "unchecked" }) @Test public void testDoStartFailsIfFileCannotBeCreated(final VertxTestContext ctx) { registrationConfig.setSaveToFile(true); when(fileSystem.existsBlocking(registrationConfig.getFilename())).thenReturn(Boolean.FALSE); doAnswer(invocation -> { final Handler<AsyncResult<Void>> handler = invocation.getArgument(1); handler.handle(Future.failedFuture("no access")); return null; }).when(fileSystem).createFile(eq(registrationConfig.getFilename()), any(Handler.class)); final Promise<Void> startupTracker = Promise.promise(); startupTracker.future().onComplete(ctx.failing(started -> { ctx.completeNow(); })); registrationService.start().onComplete(startupTracker); }
@SuppressWarnings({ "unchecked" }) @Test public void testDoStartIgnoresMalformedJson(final VertxTestContext ctx) { when(fileSystem.existsBlocking(registrationConfig.getFilename())).thenReturn(Boolean.TRUE); doAnswer(invocation -> { final Handler<AsyncResult<Buffer>> handler = invocation.getArgument(1); handler.handle(Future.succeededFuture(Buffer.buffer("NO JSON"))); return null; }).when(fileSystem).readFile(eq(registrationConfig.getFilename()), any(Handler.class)); final Promise<Void> startupTracker = Promise.promise(); startupTracker.future().onComplete(ctx.succeeding(started -> { ctx.completeNow(); })); registrationService.start().onComplete(startupTracker); }
@SuppressWarnings("unchecked") @Test public void testDoStartLoadsDeviceIdentities(final VertxTestContext ctx) { when(fileSystem.existsBlocking(registrationConfig.getFilename())).thenReturn(Boolean.TRUE); doAnswer(invocation -> { final Buffer data = DeviceRegistryTestUtils.readFile(FILE_NAME); final Handler<AsyncResult<Buffer>> handler = invocation.getArgument(1); handler.handle(Future.succeededFuture(data)); return null; }).when(fileSystem).readFile(eq(registrationConfig.getFilename()), any(Handler.class)); final Promise<Void> startupTracker = Promise.promise(); startupTracker.future() .compose(ok -> assertCanReadDevice(TENANT, DEVICE)) .compose(ok -> assertCanReadDevice(TENANT, GW)) .compose(ok -> assertDevice(TENANT, "4712", Optional.of(GW), r -> { assertEquals(HttpURLConnection.HTTP_OK, r.getStatus()); assertNotNull(r.getPayload()); assertEquals(Collections.singletonList(GW), r.getPayload().getVia()); }, r -> { assertEquals(HttpURLConnection.HTTP_OK, r.getStatus()); })) .compose(ok -> assertDevice(TENANT, "4713", Optional.of(GW), r -> { assertEquals(HttpURLConnection.HTTP_OK, r.getStatus()); assertNotNull(r.getPayload()); assertEquals(Collections.singletonList(GW), r.getPayload().getVia()); }, r -> { assertEquals(HttpURLConnection.HTTP_OK, r.getStatus()); }) .onComplete(ctx.succeeding(s -> ctx.completeNow()))); registrationService.start().onComplete(startupTracker); }
@SuppressWarnings({ "unchecked" }) @Test public void testDoStartIgnoreIdentitiesIfStartEmptyIsSet(final VertxTestContext ctx) { registrationConfig.setFilename(FILE_NAME); registrationConfig.setStartEmpty(true); when(fileSystem.existsBlocking(registrationConfig.getFilename())).thenReturn(Boolean.TRUE); final Promise<Void> startupTracker = Promise.promise(); startupTracker.future().onComplete(ctx.succeeding(s -> ctx.verify(() -> { verify(fileSystem, never()).readFile(anyString(), any(Handler.class)); ctx.completeNow(); }))); registrationService.start().onComplete(startupTracker); }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testPeriodicSafeJobIsNotScheduledIfSavingIfDisabled(final VertxTestContext ctx) { registrationConfig.setSaveToFile(false); when(fileSystem.existsBlocking(registrationConfig.getFilename())).thenReturn(Boolean.TRUE); doAnswer(invocation -> { final Handler handler = invocation.getArgument(1); handler.handle(Future.failedFuture("malformed file")); return null; }).when(fileSystem).readFile(eq(registrationConfig.getFilename()), any(Handler.class)); final Promise<Void> startupTracker = Promise.promise(); startupTracker.future().onComplete(ctx.succeeding(done -> ctx.verify(() -> { verify(vertx, never()).setPeriodic(anyLong(), any(Handler.class)); ctx.completeNow(); }))); registrationService.start().onComplete(startupTracker); }