PubSubRecordItemListener implements RecordItemListener { @Override public void onItem(RecordItem recordItem) throws ImporterException { TransactionBody body = recordItem.getTransactionBody(); TransactionRecord txRecord = recordItem.getRecord(); TransactionHandler transactionHandler = transactionHandlerFactory.create(body); log.trace("Storing transaction body: {}", () -> Utility.printProtoMessage(body)); long consensusTimestamp = Utility.timeStampInNanos(txRecord.getConsensusTimestamp()); EntityId entity = transactionHandler.getEntity(recordItem); PubSubMessage pubSubMessage = buildPubSubMessage(consensusTimestamp, entity, recordItem); try { sendPubSubMessage(pubSubMessage); } catch (Exception e) { throw new ParserException("Error sending transaction to pubsub", e); } log.debug("Published transaction : {}", consensusTimestamp); if (addressBookService.isAddressBook(entity)) { FileID fileID = null; byte[] fileBytes = null; if (body.hasFileAppend()) { fileID = body.getFileAppend().getFileID(); fileBytes = body.getFileAppend().getContents().toByteArray(); } else if (body.hasFileCreate()) { fileID = txRecord.getReceipt().getFileID(); fileBytes = body.getFileCreate().getContents().toByteArray(); } else if (body.hasFileUpdate()) { fileID = body.getFileUpdate().getFileID(); fileBytes = body.getFileUpdate().getContents().toByteArray(); } FileData fileData = new FileData(consensusTimestamp, fileBytes, EntityId.of(fileID), recordItem .getTransactionType()); fileDataRepository.save(fileData); addressBookService.update(fileData); } }  @Override void onItem(RecordItem recordItem);  }
@Test void testPubSubMessage() throws Exception { byte[] message = new byte[] {'a', 'b', 'c'}; TopicID topicID = TopicID.newBuilder().setTopicNum(10L).build(); EntityId topicIdEntity = EntityId.of(topicID); ConsensusSubmitMessageTransactionBody submitMessage = ConsensusSubmitMessageTransactionBody.newBuilder() .setMessage(ByteString.copyFrom(message)) .setTopicID(topicID) .build(); Transaction transaction = buildTransaction(builder -> builder.setConsensusSubmitMessage(submitMessage)); doReturn(topicIdEntity).when(transactionHandler).getEntity(any()); pubSubRecordItemListener.onItem(new RecordItem(transaction.toByteArray(), DEFAULT_RECORD_BYTES)); var pubSubMessage = assertPubSubMessage(buildPubSubTransaction(transaction), 1); assertThat(pubSubMessage.getEntity()).isEqualTo(topicIdEntity); assertThat(pubSubMessage.getNonFeeTransfers()).isNull(); }
@Test void testPubSubMessageWithNonFeeTransferAndNullEntityId() throws Exception { List<AccountAmount> nonFeeTransfers = new ArrayList<>(); nonFeeTransfers.add(buildAccountAmount(10L, 100L)); nonFeeTransfers.add(buildAccountAmount(11L, 111L)); CryptoTransferTransactionBody cryptoTransfer = CryptoTransferTransactionBody.newBuilder() .setTransfers(TransferList.newBuilder() .addAccountAmounts(nonFeeTransfers.get(0)) .addAccountAmounts(nonFeeTransfers.get(1)) .build()) .build(); Transaction transaction = buildTransaction(builder -> builder.setCryptoTransfer(cryptoTransfer)); pubSubRecordItemListener.onItem(new RecordItem(transaction.toByteArray(), DEFAULT_RECORD_BYTES)); var pubSubMessage = assertPubSubMessage(buildPubSubTransaction(transaction), 1); assertThat(pubSubMessage.getEntity()).isNull(); assertThat(pubSubMessage.getNonFeeTransfers()).isEqualTo(nonFeeTransfers); }
@Test void testNonRetryableError() { CryptoTransferTransactionBody cryptoTransfer = CryptoTransferTransactionBody.newBuilder() .setTransfers(TransferList.newBuilder().build()) .build(); Transaction transaction = buildTransaction(builder -> builder.setCryptoTransfer(cryptoTransfer)); when(messageChannel.send(any())).thenThrow(RuntimeException.class); assertThatThrownBy( () -> pubSubRecordItemListener.onItem( new RecordItem(transaction.toByteArray(), DEFAULT_RECORD_BYTES))) .isInstanceOf(ParserException.class) .hasMessageContaining("Error sending transaction to pubsub"); verify(messageChannel, times(1)).send(any()); }
@Test void testSendRetries() throws Exception { CryptoTransferTransactionBody cryptoTransfer = CryptoTransferTransactionBody.newBuilder() .setTransfers(TransferList.newBuilder().build()) .build(); Transaction transaction = buildTransaction(builder -> builder.setCryptoTransfer(cryptoTransfer)); pubSubProperties.setMaxSendAttempts(3); when(messageChannel.send(any())) .thenThrow(MessageTimeoutException.class) .thenThrow(MessageTimeoutException.class) .thenReturn(true); pubSubRecordItemListener.onItem(new RecordItem(transaction.toByteArray(), DEFAULT_RECORD_BYTES)); var pubSubMessage = assertPubSubMessage(buildPubSubTransaction(transaction), 3); assertThat(pubSubMessage.getEntity()).isNull(); assertThat(pubSubMessage.getNonFeeTransfers()).isNull(); }
@Test void testNetworkAddressBookAppend() throws Exception { byte[] fileContents = new byte[] {0b0, 0b1, 0b10}; Transaction transaction = buildTransaction(builder -> { builder.setFileAppend(FileAppendTransactionBody.newBuilder() .setFileID(ADDRESS_BOOK_FILE_ID) .setContents(ByteString.copyFrom(fileContents)) .build()); }); EntityId entityId = EntityId.of(ADDRESS_BOOK_FILE_ID); doReturn(EntityId.of(ADDRESS_BOOK_FILE_ID)).when(transactionHandler).getEntity(any()); pubSubRecordItemListener.onItem(new RecordItem(transaction.toByteArray(), DEFAULT_RECORD_BYTES)); FileData fileData = new FileData(100L, fileContents, entityId, TransactionTypeEnum.FILEAPPEND .getProtoId()); verify(addressBookService).update(fileData); }
@Test void testNetworkAddressBookUpdate() throws Exception { byte[] fileContents = UPDATED.toByteArray(); var fileUpdate = FileUpdateTransactionBody.newBuilder() .setFileID(ADDRESS_BOOK_FILE_ID) .setContents(ByteString.copyFrom(fileContents)) .build(); Transaction transaction = buildTransaction(builder -> builder.setFileUpdate(fileUpdate)); EntityId entityId = EntityId.of(ADDRESS_BOOK_FILE_ID); doReturn(EntityId.of(ADDRESS_BOOK_FILE_ID)).when(transactionHandler).getEntity(any()); pubSubRecordItemListener.onItem(new RecordItem(transaction.toByteArray(), DEFAULT_RECORD_BYTES)); FileData fileData = new FileData(100L, fileContents, EntityId .of(ADDRESS_BOOK_FILE_ID), TransactionTypeEnum.FILEUPDATE .getProtoId()); verify(addressBookService).update(fileData); }