@Value.Immutable
@Gson.TypeAdapters
public interface FileData {
+ String productName();
+
+ String vendorName();
+
+ String lastEpochMicrosec();
+
+ String sourceName();
+
+ String startEpochMicrosec();
+
+ String timeZoneOffset();
+
String changeIdentifier();
String changeType();
public class DmaapConsumerJsonParser {
private static final Logger logger = LoggerFactory.getLogger(DmaapConsumerJsonParser.class);
+ private static final String COMMON_EVENT_HEADER = "commonEventHeader";
+ private static final String EVENT_NAME = "eventName";
+ private static final String LAST_EPOCH_MICROSEC = "lastEpochMicrosec";
+ private static final String SOURCE_NAME = "sourceName";
+ private static final String START_EPOCH_MICROSEC = "startEpochMicrosec";
+ private static final String TIME_ZONE_OFFSET = "timeZoneOffset";
+
private static final String EVENT = "event";
private static final String NOTIFICATION_FIELDS = "notificationFields";
private static final String CHANGE_IDENTIFIER = "changeIdentifier";
private Flux<FileData> transform(JsonObject jsonObject) {
if (containsHeader(jsonObject, EVENT, NOTIFICATION_FIELDS)) {
+ JsonObject commonEventHeader = jsonObject.getAsJsonObject(EVENT).getAsJsonObject(COMMON_EVENT_HEADER);
+ String eventName = getValueFromJson(commonEventHeader, EVENT_NAME);
+ String productName = getProductNameFromEventName(eventName);
+ String vendorName = getVendorNameFromEventName(eventName);
+ String lastEpochMicrosec = getValueFromJson(commonEventHeader, LAST_EPOCH_MICROSEC);
+ String sourceName = getValueFromJson(commonEventHeader, SOURCE_NAME);
+ String startEpochMicrosec = getValueFromJson(commonEventHeader, START_EPOCH_MICROSEC);
+ String timeZoneOffset = getValueFromJson(commonEventHeader, TIME_ZONE_OFFSET);
+
JsonObject notificationFields = jsonObject.getAsJsonObject(EVENT).getAsJsonObject(NOTIFICATION_FIELDS);
String changeIdentifier = getValueFromJson(notificationFields, CHANGE_IDENTIFIER);
String changeType = getValueFromJson(notificationFields, CHANGE_TYPE);
if (isNotificationFieldsHeaderNotEmpty(changeIdentifier, changeType, notificationFieldsVersion)
&& arrayOfNamedHashMap != null && isChangeIdentifierCorrect(changeIdentifier)
&& isChangeTypeCorrect(changeType)) {
- return getAllFileDataFromJson(changeIdentifier, changeType, arrayOfNamedHashMap);
+ return getAllFileDataFromJson(productName, vendorName, lastEpochMicrosec, sourceName,
+ startEpochMicrosec, timeZoneOffset, changeIdentifier, changeType, arrayOfNamedHashMap);
}
return handleJsonError(changeIdentifier, changeType, notificationFieldsVersion, arrayOfNamedHashMap,
return FILE_READY_CHANGE_IDENTIFIER.equals(changeIdentifier);
}
- private Flux<FileData> getAllFileDataFromJson(String changeIdentifier, String changeType,
- JsonArray arrayOfAdditionalFields) {
+ private Flux<FileData> getAllFileDataFromJson(String productName, String vendorName, String lastEpochMicrosec,
+ String sourceName, String startEpochMicrosec, String timeZoneOffset, String changeIdentifier,
+ String changeType, JsonArray arrayOfAdditionalFields) {
List<FileData> res = new ArrayList<>();
for (int i = 0; i < arrayOfAdditionalFields.size(); i++) {
if (arrayOfAdditionalFields.get(i) != null) {
JsonObject fileInfo = (JsonObject) arrayOfAdditionalFields.get(i);
- FileData fileData = getFileDataFromJson(fileInfo, changeIdentifier, changeType);
+ FileData fileData = getFileDataFromJson(productName, vendorName, lastEpochMicrosec, sourceName,
+ startEpochMicrosec, timeZoneOffset, fileInfo, changeIdentifier, changeType);
if (fileData != null) {
res.add(fileData);
return Flux.fromIterable(res);
}
- private FileData getFileDataFromJson(JsonObject fileInfo, String changeIdentifier, String changeType) {
+ private FileData getFileDataFromJson(String productName, String vendorName, String lastEpochMicrosec,
+ String sourceName, String startEpochMicrosec, String timeZoneOffset, JsonObject fileInfo,
+ String changeIdentifier, String changeType) {
logger.trace("starting to getFileDataFromJson!");
FileData fileData = null;
if (isFileFormatFieldsNotEmpty(fileFormatVersion, fileFormatType)
&& isNameAndLocationAndCompressionNotEmpty(name, location, compression)) {
- fileData = ImmutableFileData.builder().name(name).changeIdentifier(changeIdentifier).changeType(changeType)
- .location(location).compression(compression).fileFormatType(fileFormatType)
- .fileFormatVersion(fileFormatVersion).build();
+ // @formatter:off
+ fileData = ImmutableFileData.builder()
+ .productName(productName)
+ .vendorName(vendorName)
+ .lastEpochMicrosec(lastEpochMicrosec)
+ .sourceName(sourceName)
+ .startEpochMicrosec(startEpochMicrosec)
+ .timeZoneOffset(timeZoneOffset)
+ .name(name)
+ .changeIdentifier(changeIdentifier)
+ .changeType(changeType)
+ .location(location)
+ .compression(compression)
+ .fileFormatType(fileFormatType)
+ .fileFormatVersion(fileFormatVersion)
+ .build();
+ // @formatter:on
}
return fileData;
}
+ /**
+ * @param eventName
+ * @return String of vendorName eventName is defined as:
+ * {DomainAbbreviation}_{productName}-{vendorName}_{Description}, example:
+ * Noti_RnNode-Ericsson_FileReady
+ */
+ private String getVendorNameFromEventName(String eventName) {
+ String[] eventArray = eventName.split("_|-");
+ if (eventArray.length >= 4) {
+ return eventArray[2];
+ } else {
+ logger.trace("Can not get vendorName from eventName, eventName is not in correct format: " + eventName);
+ }
+ return "";
+ }
+
+ /**
+ * @param eventName
+ * @return String of productName
+ */
+ private String getProductNameFromEventName(String eventName) {
+ String[] eventArray = eventName.split("_|-");
+ if (eventArray.length >= 4) {
+ return eventArray[1];
+ } else {
+ logger.trace("Can not get productName from eventName, eventName is not in correct format: " + eventName);
+ }
+ return "";
+ }
+
private String getValueFromJson(JsonObject jsonObject, String jsonKey) {
return jsonObject.has(jsonKey) ? jsonObject.get(jsonKey).getAsString() : "";
}
private FileServerData getFileServerData(URI uri) {
String[] userInfo = getUserNameAndPasswordIfGiven(uri.getUserInfo());
- return ImmutableFileServerData.builder().serverAddress(uri.getHost())
- .userId(userInfo != null ? userInfo[0] : "").password(userInfo != null ? userInfo[1] : "")
- .port(uri.getPort()).build();
+ // @formatter:off
+ return ImmutableFileServerData.builder()
+ .serverAddress(uri.getHost())
+ .userId(userInfo != null ? userInfo[0] : "")
+ .password(userInfo != null ? userInfo[1] : "")
+ .port(uri.getPort())
+ .build();
+ // @formatter:on
}
private String[] getUserNameAndPasswordIfGiven(String userInfoString) {
}
private ConsumerDmaapModel getConsumerDmaapModel(FileData fileData, String localFile) {
+ String productName = fileData.productName();
+ String vendorName = fileData.vendorName();
+ String lastEpochMicrosec = fileData.lastEpochMicrosec();
+ String sourceName = fileData.sourceName();
+ String startEpochMicrosec = fileData.startEpochMicrosec();
+ String timeZoneOffset = fileData.timeZoneOffset();
String name = fileData.name();
+ String location = fileData.location();
+ String internalLocation = localFile;
String compression = fileData.compression();
String fileFormatType = fileData.fileFormatType();
String fileFormatVersion = fileData.fileFormatVersion();
- return ImmutableConsumerDmaapModel.builder().name(name).location(localFile).compression(compression)
- .fileFormatType(fileFormatType).fileFormatVersion(fileFormatVersion).build();
+ // @formatter:off
+ return ImmutableConsumerDmaapModel.builder()
+ .productName(productName)
+ .vendorName(vendorName)
+ .lastEpochMicrosec(lastEpochMicrosec)
+ .sourceName(sourceName)
+ .startEpochMicrosec(startEpochMicrosec)
+ .timeZoneOffset(timeZoneOffset)
+ .name(name)
+ .location(location)
+ .internalLocation(internalLocation)
+ .compression(compression)
+ .fileFormatType(fileFormatType)
+ .fileFormatVersion(fileFormatVersion)
+ .build();
+ // @formatter:on
}
private RetryTimer getRetryTimer() {
* @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
*/
class DmaapConsumerJsonParserTest {
+ private static final String PRODUCT_NAME = "NrRadio";
+ private static final String VENDOR_NAME = "Ericsson";
+ private static final String LAST_EPOCH_MICROSEC = "1519837825682";
+ private static final String SOURCE_NAME = "5GRAN_DU";
+ private static final String START_EPOCH_MICROSEC = "1519837825682";
+ private static final String TIME_ZONE_OFFSET = "UTC+05:00";
private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
private static final String LOCATION = "ftpes://192.168.0.101:22/ftp/rop/" + PM_FILE_NAME;
private static final String GZIP_COMPRESSION = "gzip";
@Test
void whenPassingCorrectJson_validationNotThrowingAnException() throws DmaapNotFoundException {
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME).location(LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION)
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalField)
.build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalField).build();
-
- FileData expectedFileData = ImmutableFileData.builder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).name(PM_FILE_NAME).location(LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+ FileData expectedFileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingCorrectJsonWithoutName_noFileData() {
- AdditionalField additionalField =
- new JsonMessage.AdditionalFieldBuilder().location(LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalField).build();
-
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalField)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingCorrectJsonWithoutLocation_noFileData() {
- AdditionalField additionalField =
- new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME).compression(GZIP_COMPRESSION)
- .fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalField).build();
-
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalField)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingCorrectJsonWithoutCompression_noFileData() {
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME).location(LOCATION)
- .fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalField).build();
-
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalField)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingCorrectJsonWithoutFileFormatType_noFileData() {
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME).location(LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalField).build();
-
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalField)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingOneCorrectJsonWithoutFileFormatVersionAndOneCorrect_oneFileData() {
- AdditionalField additionalFaultyField = new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME)
- .location(LOCATION).compression(GZIP_COMPRESSION).fileFormatType(FILE_FORMAT_TYPE).build();
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME).location(LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION)
+ // @formatter:off
+ AdditionalField additionalFaultyField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .build();
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalFaultyField)
+ .addAdditionalField(additionalField)
.build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalFaultyField).addAdditionalField(additionalField).build();
-
- FileData expectedFileData = ImmutableFileData.builder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).name(PM_FILE_NAME).location(LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+ FileData expectedFileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingJsonWithoutMandatoryHeaderInformation_validationThrowingAnException() {
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES_INVALID")
- .changeType("FileReady_INVALID").notificationFieldsVersion("1.0_INVALID").build();
-
+ // @formatter:off
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier("PM_MEAS_FILES_INVALID")
+ .changeType("FileReady_INVALID")
+ .notificationFieldsVersion("1.0_INVALID")
+ .build();
+ // @formatter:on
String incorrectMessageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingCorrectJsonWithIncorrectChangeType_validationThrowingAnException() {
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME).location(LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(CHANGE_IDENTIFIER)
- .changeType(INCORRECT_CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalField).build();
-
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(CHANGE_IDENTIFIER)
+ .changeType(INCORRECT_CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalField)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@Test
void whenPassingCorrectJsonWithIncorrectChangeIdentifier_validationThrowingAnException() {
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().name(PM_FILE_NAME).location(LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier(INCORRECT_CHANGE_IDENTIFIER)
- .changeType(CHANGE_TYPE).notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
- .addAdditionalField(additionalField).build();
-
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name(PM_FILE_NAME)
+ .location(LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(INCORRECT_CHANGE_IDENTIFIER)
+ .changeType(CHANGE_TYPE)
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION)
+ .addAdditionalField(additionalField)
+ .build();
+ // @formatter:on
String messageString = message.toString();
String parsedString = message.getParsed();
DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
* @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
*/
class DmaapConsumerTaskImplTest {
+ private static final String PRODUCT_NAME = "NrRadio";
+ private static final String VENDOR_NAME = "Ericsson";
+ private static final String LAST_EPOCH_MICROSEC = "8745745764578";
+ private static final String SOURCE_NAME = "oteNB5309";
+ private static final String START_EPOCH_MICROSEC = "8745745764578";
+ private static final String TIME_ZONE_OFFSET = "UTC+05:00";
private static final String PM_MEAS_CHANGE_IDENTIFIER = "PM_MEAS_FILES";
private static final String FILE_READY_CHANGE_TYPE = "FileReady";
private static final String FTPES_SCHEME = "ftpes://";
@BeforeAll
public static void setUp() {
- dmaapConsumerConfiguration = new ImmutableDmaapConsumerConfiguration.Builder().consumerGroup("OpenDCAE-c12")
- .consumerId("c12").dmaapContentType("application/json").dmaapHostName("54.45.33.2")
- .dmaapPortNumber(1234).dmaapProtocol("https").dmaapUserName("Datafile").dmaapUserPassword("Datafile")
- .dmaapTopicName("unauthenticated.NOTIFICATION").timeoutMS(-1).messageLimit(-1).build();
+ //@formatter:off
+ dmaapConsumerConfiguration = new ImmutableDmaapConsumerConfiguration.Builder()
+ .consumerGroup("OpenDCAE-c12")
+ .consumerId("c12")
+ .dmaapContentType("application/json")
+ .dmaapHostName("54.45.33.2")
+ .dmaapPortNumber(1234).dmaapProtocol("https")
+ .dmaapUserName("Datafile")
+ .dmaapUserPassword("Datafile")
+ .dmaapTopicName("unauthenticated.NOTIFICATION")
+ .timeoutMS(-1)
+ .messageLimit(-1)
+ .build();
appConfig = mock(AppConfig.class);
- AdditionalField ftpesAdditionalField =
- new JsonMessage.AdditionalFieldBuilder().location(FTPES_LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage ftpesJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
- .addAdditionalField(ftpesAdditionalField).build();
- ftpesMessage = ftpesJsonMessage.toString();
- ftpesFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(FTPES_LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
- .fileFormatVersion(FILE_FORMAT_VERSION).build();
-
- AdditionalField sftpAdditionalField =
- new JsonMessage.AdditionalFieldBuilder().location(SFTP_LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
- JsonMessage sftpJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
- .addAdditionalField(sftpAdditionalField).build();
- sftpMessage = sftpJsonMessage.toString();
- sftpFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(SFTP_LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
- .fileFormatVersion(FILE_FORMAT_VERSION).build();
+ AdditionalField ftpesAdditionalField = new JsonMessage.AdditionalFieldBuilder()
+ .location(FTPES_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage ftpesJsonMessage = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .notificationFieldsVersion("1.0")
+ .addAdditionalField(ftpesAdditionalField)
+ .build();
- ImmutableConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder().name(PM_FILE_NAME)
- .location(LOCAL_FILE_LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+ ftpesMessage = ftpesJsonMessage.toString();
+ ftpesFileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(FTPES_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+
+ AdditionalField sftpAdditionalField = new JsonMessage.AdditionalFieldBuilder()
+ .location(SFTP_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ JsonMessage sftpJsonMessage = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .notificationFieldsVersion("1.0")
+ .addAdditionalField(sftpAdditionalField)
+ .build();
+ sftpMessage = sftpJsonMessage.toString();
+ sftpFileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(SFTP_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+
+
+ ImmutableConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .name(PM_FILE_NAME)
+ .location(FTPES_LOCATION)
+ .internalLocation(LOCAL_FILE_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
listOfConsumerDmaapModel.add(consumerDmaapModel);
+ //@formatter:on
}
@Test
* @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
*/
class DmaapPublisherTaskImplTest {
+ private static final String PRODUCT_NAME = "NrRadio";
+ private static final String VENDOR_NAME = "Ericsson";
+ private static final String LAST_EPOCH_MICROSEC = "8745745764578";
+ private static final String SOURCE_NAME = "oteNB5309";
+ private static final String START_EPOCH_MICROSEC = "8745745764578";
+ private static final String TIME_ZONE_OFFSET = "UTC+05:00";
private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
private static ConsumerDmaapModel consumerDmaapModel;
@BeforeAll
public static void setUp() {
- dmaapPublisherConfiguration =
- new ImmutableDmaapPublisherConfiguration.Builder().dmaapContentType("application/json")
- .dmaapHostName("54.45.33.2").dmaapPortNumber(1234).dmaapProtocol("https").dmaapUserName("DFC")
- .dmaapUserPassword("DFC").dmaapTopicName("unauthenticated.VES_NOTIFICATION_OUTPUT").build();
- consumerDmaapModel = ImmutableConsumerDmaapModel.builder().name(PM_FILE_NAME).location("target/" + PM_FILE_NAME)
- .compression("gzip").fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
+ //@formatter:off
+ dmaapPublisherConfiguration = new ImmutableDmaapPublisherConfiguration.Builder()
+ .dmaapContentType("application/json")
+ .dmaapHostName("54.45.33.2")
+ .dmaapPortNumber(1234)
+ .dmaapProtocol("https")
+ .dmaapUserName("DFC")
+ .dmaapUserPassword("DFC")
+ .dmaapTopicName("unauthenticated.VES_NOTIFICATION_OUTPUT")
+ .build();
+ consumerDmaapModel = ImmutableConsumerDmaapModel.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .name(PM_FILE_NAME)
+ .location("ftpes://192.168.0.101:22/ftp/rop/" + PM_FILE_NAME)
+ .internalLocation("target/" + PM_FILE_NAME)
+ .compression("gzip")
+ .fileFormatType("org.3GPP.32.435#measCollec")
+ .fileFormatVersion("V10")
+ .build();
appConfig = mock(AppConfig.class);
+ //@formatter:on
}
@Test
*
*/
public class XnfCollectorTaskImplTest {
-
+ private static final String PRODUCT_NAME = "NrRadio";
+ private static final String VENDOR_NAME = "Ericsson";
+ private static final String LAST_EPOCH_MICROSEC = "8745745764578";
+ private static final String SOURCE_NAME = "oteNB5309";
+ private static final String START_EPOCH_MICROSEC = "8745745764578";
+ private static final String TIME_ZONE_OFFSET = "UTC+05:00";
private static final String PM_MEAS_CHANGE_IDINTIFIER = "PM_MEAS_FILES";
private static final String FILE_READY_CHANGE_TYPE = "FileReady";
private static final String FTPES_SCHEME = "ftpes://";
@Test
public void whenFtpesFile_returnCorrectResponse() {
- XnfCollectorTaskImpl collectorUndetTest = new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
-
- FileData fileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(FTPES_LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
- .fileFormatVersion(FILE_FORMAT_VERSION).build();
-
- FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS).userId(USER)
- .password(PWD).port(PORT_22).build();
+ XnfCollectorTaskImpl collectorUndetTest =
+ new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
+
+ // @formatter:off
+ FileData fileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(FTPES_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+
+ FileServerData fileServerData = ImmutableFileServerData.builder()
+ .serverAddress(SERVER_ADDRESS)
+ .userId(USER)
+ .password(PWD)
+ .port(PORT_22)
+ .build();
+ // @formatter:on
when(ftpsClientMock.collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION))
.thenReturn(new FileCollectResult());
- ConsumerDmaapModel expectedConsumerDmaapModel = ImmutableConsumerDmaapModel.builder().name(PM_FILE_NAME)
- .location(LOCAL_FILE_LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+ // @formatter:off
+ ConsumerDmaapModel expectedConsumerDmaapModel = ImmutableConsumerDmaapModel.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .name(PM_FILE_NAME)
+ .location(FTPES_LOCATION)
+ .internalLocation(LOCAL_FILE_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ // @formatter:on
StepVerifier.create(collectorUndetTest.execute(fileData)).expectNext(expectedConsumerDmaapModel)
.verifyComplete();
@Test
public void whenSftpFile_returnCorrectResponse() {
- XnfCollectorTaskImpl collectorUndetTest = new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
-
- FileData fileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(SFTP_LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
- .fileFormatVersion(FILE_FORMAT_VERSION).build();
-
- FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS).userId("")
- .password("").port(PORT_22).build();
+ XnfCollectorTaskImpl collectorUndetTest =
+ new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
+ // @formatter:off
+ FileData fileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(SFTP_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+
+
+ FileServerData fileServerData = ImmutableFileServerData.builder()
+ .serverAddress(SERVER_ADDRESS)
+ .userId("")
+ .password("")
+ .port(PORT_22)
+ .build();
+ // @formatter:on
when(sftpClientMock.collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION))
.thenReturn(new FileCollectResult());
-
- ConsumerDmaapModel expectedConsumerDmaapModel = ImmutableConsumerDmaapModel.builder().name(PM_FILE_NAME)
- .location(LOCAL_FILE_LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
-
+ // @formatter:off
+ ConsumerDmaapModel expectedConsumerDmaapModel = ImmutableConsumerDmaapModel.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .name(PM_FILE_NAME)
+ .location(SFTP_LOCATION)
+ .internalLocation(LOCAL_FILE_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ // @formatter:on
StepVerifier.create(collectorUndetTest.execute(fileData)).expectNext(expectedConsumerDmaapModel)
.verifyComplete();
@Test
public void whenFtpesFileAlwaysFail_retryAndReturnEmpty() {
- XnfCollectorTaskImpl collectorUndetTest = new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
+ XnfCollectorTaskImpl collectorUndetTest =
+ new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
collectorUndetTest.setRetryTimer(retryTimerMock);
-
- FileData fileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(FTPES_LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
- .fileFormatVersion(FILE_FORMAT_VERSION).build();
-
- FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS).userId(USER)
- .password(PWD).port(PORT_22).build();
+ // @formatter:off
+ FileData fileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(FTPES_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+
+ FileServerData fileServerData = ImmutableFileServerData.builder()
+ .serverAddress(SERVER_ADDRESS)
+ .userId(USER)
+ .password(PWD)
+ .port(PORT_22)
+ .build();
+ // @formatter:on
ErrorData errorData = new ErrorData();
errorData.addError("Unable to collect file.", new Exception());
when(ftpsClientMock.collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION))
@Test
public void whenFtpesFileFailOnce_retryAndReturnCorrectResponse() {
- XnfCollectorTaskImpl collectorUndetTest = new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
+ XnfCollectorTaskImpl collectorUndetTest =
+ new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
collectorUndetTest.setRetryTimer(retryTimerMock);
-
- FileData fileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(FTPES_LOCATION)
- .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
- .fileFormatVersion(FILE_FORMAT_VERSION).build();
-
- FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS).userId(USER)
- .password(PWD).port(PORT_22).build();
+ // @formatter:off
+ FileData fileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location(FTPES_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+
+ FileServerData fileServerData = ImmutableFileServerData.builder()
+ .serverAddress(SERVER_ADDRESS)
+ .userId(USER)
+ .password(PWD)
+ .port(PORT_22)
+ .build();
+ // @formatter:on
ErrorData errorData = new ErrorData();
errorData.addError("Unable to collect file.", new Exception());
when(ftpsClientMock.collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION))
- .thenReturn(new FileCollectResult(errorData));
+ .thenReturn(new FileCollectResult(errorData));
doReturn(new FileCollectResult()).when(ftpsClientMock).retryCollectFile();
-
- ConsumerDmaapModel expectedConsumerDmaapModel = ImmutableConsumerDmaapModel.builder().name(PM_FILE_NAME)
- .location(LOCAL_FILE_LOCATION).compression(GZIP_COMPRESSION)
- .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
-
+ // @formatter:off
+ ConsumerDmaapModel expectedConsumerDmaapModel = ImmutableConsumerDmaapModel.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .name(PM_FILE_NAME)
+ .location(FTPES_LOCATION)
+ .internalLocation(LOCAL_FILE_LOCATION)
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ // @formatter:on
StepVerifier.create(collectorUndetTest.execute(fileData)).expectNext(expectedConsumerDmaapModel)
.verifyComplete();
@Test
public void whenWrongScheme_returnEmpty() {
- XnfCollectorTaskImpl collectorUndetTest = new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
-
- FileData fileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
- .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location("http://host.com/file.zip")
- .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
- .fileFormatVersion(FILE_FORMAT_VERSION).build();
-
- FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS).userId("")
- .password("").port(PORT_22).build();
+ XnfCollectorTaskImpl collectorUndetTest =
+ new XnfCollectorTaskImpl(appConfigMock, ftpsClientMock, sftpClientMock);
+ // @formatter:off
+ FileData fileData = ImmutableFileData.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+ .changeType(FILE_READY_CHANGE_TYPE)
+ .name(PM_FILE_NAME)
+ .location("http://host.com/file.zip")
+ .compression(GZIP_COMPRESSION)
+ .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+
+ FileServerData fileServerData = ImmutableFileServerData.builder()
+ .serverAddress(SERVER_ADDRESS)
+ .userId("")
+ .password("")
+ .port(PORT_22)
+ .build();
+ // @formatter:on
when(sftpClientMock.collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION))
.thenReturn(new FileCollectResult());
}
additionalFieldsString.append("]");
}
-
- return "{" + "\"event\":{" + "\"commonEventHeader\":{" + "\"domain\":\"notification\","
- + "\"eventId\":\"<<SerialNumber>>-reg\"," + "\"eventName\":\"EriNoti_RnNode_FileReady\","
- + "\"eventType\":\"fileReady\"," + "\"internalHeaderFields\":{},"
- + "\"lastEpochMicrosec\":1519837825682," + "\"nfNamingCode\":\"5GRAN\"," + "\"nfcNamingCode\":\"5DU\","
- + "\"priority\":\"Normal\"," + "\"reportingEntityName\":\"5GRAN_DU\"," + "\"sequence\":0,"
- + "\"sourceId\":\"<<SerialNumber>>\"," + "\"sourceName\":\"5GRAN_DU\","
- + "\"startEpochMicrosec\":\"1519837825682\"," + "\"version\":3" + "}," + "\"notificationFields\":{"
+ // @formatter:off
+ return "{" + "\"event\":{"
+ + "\"commonEventHeader\":{"
+ + "\"domain\":\"notification\","
+ + "\"eventId\":\"<<SerialNumber>>-reg\","
+ + "\"eventName\":\"Noti_NrRadio-Ericsson_FileReady\","
+ + "\"eventType\":\"fileReady\","
+ + "\"internalHeaderFields\":{},"
+ + "\"lastEpochMicrosec\":1519837825682,"
+ + "\"nfNamingCode\":\"5GRAN\","
+ + "\"nfcNamingCode\":\"5DU\","
+ + "\"priority\":\"Normal\","
+ + "\"reportingEntityName\":\"5GRAN_DU\","
+ + "\"sequence\":0,"
+ + "\"sourceId\":\"<<SerialNumber>>\","
+ + "\"sourceName\":\"5GRAN_DU\","
+ + "\"timeZoneOffset\":\"UTC+05:00\","
+ + "\"startEpochMicrosec\":\"1519837825682\","
+ + "\"version\":3"
+ + "},"
+ + "\"notificationFields\":{"
+ // @formatter:on
+ getAsStringIfParameterIsSet("changeIdentifier", changeIdentifier,
changeType != null || notificationFieldsVersion != null || arrayOfAdditionalFields.size() > 0)
+ getAsStringIfParameterIsSet("changeType", changeType,
@Override
public String toString() {
- return "{" + getAsStringIfParameterIsSet("name", name, true)
- + "\"hashMap\":{"
+ return "{" + getAsStringIfParameterIsSet("name", name, true) + "\"hashMap\":{"
+ getAsStringIfParameterIsSet("location", location,
compression != null || fileFormatType != null || fileFormatVersion != null)
+ getAsStringIfParameterIsSet("compression", compression,
* @param args Not used
*/
public static void main(String[] args) {
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().name("A20161224.1030-1045.bin.gz")
- .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz").compression("gzip")
- .fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
- AdditionalField secondAdditionalField =
- new JsonMessage.AdditionalFieldBuilder().name("A20161224.1030-1045.bin.gz")
- .location("sftp://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz").compression("gzip")
- .fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES")
- .changeType("FileReady").notificationFieldsVersion("2.0").addAdditionalField(additionalField)
- .addAdditionalField(secondAdditionalField).build();
+ // @formatter:off
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name("A20161224.1030-1045.bin.gz")
+ .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz")
+ .compression("gzip")
+ .fileFormatType("org.3GPP.32.435#measCollec")
+ .fileFormatVersion("V10")
+ .build();
+ AdditionalField secondAdditionalField = new JsonMessage.AdditionalFieldBuilder()
+ .name("A20161224.1030-1045.bin.gz")
+ .location("sftp://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz")
+ .compression("gzip")
+ .fileFormatType("org.3GPP.32.435#measCollec")
+ .fileFormatVersion("V10")
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder()
+ .changeIdentifier("PM_MEAS_FILES")
+ .changeType("FileReady")
+ .notificationFieldsVersion("2.0")
+ .addAdditionalField(additionalField)
+ .addAdditionalField(secondAdditionalField)
+ .build();
+ // @formatter:on
System.out.println(message.toString());
}
}
@Gson.TypeAdapters
public interface ConsumerDmaapModel {
+ @SerializedName("productName")
+ String getProductName();
+
+ @SerializedName("vendorName")
+ String getVendorName();
+
+ @SerializedName("lastEpochMicrosec")
+ String getLastEpochMicrosec();
+
+ @SerializedName("sourceName")
+ String getSourceName();
+
+ @SerializedName("startEpochMicrosec")
+ String getStartEpochMicrosec();
+
+ @SerializedName("timeZoneOffset")
+ String getTimeZoneOffset();
+
@SerializedName("name")
String getName();
@SerializedName("location")
String getLocation();
+ @SerializedName("internalLocation")
+ String getInternalLocation();
+
@SerializedName("compression")
String getCompression();
import org.junit.jupiter.api.Test;
class CommonFunctionsTest {
- private ConsumerDmaapModel model = ImmutableConsumerDmaapModel.builder().name("A20161224.1030-1045.bin.gz")
- .location("target/A20161224.1030-1045.bin.gz").compression("gzip")
- .fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
+ // @formatter:off
+ private ConsumerDmaapModel model = ImmutableConsumerDmaapModel.builder()
+ .productName("NrRadio")
+ .vendorName("Ericsson")
+ .lastEpochMicrosec("8745745764578")
+ .sourceName("oteNB5309")
+ .startEpochMicrosec("8745745764578")
+ .timeZoneOffset("UTC+05:00")
+ .name("A20161224.1030-1045.bin.gz")
+ .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1145.bin.gz")
+ .internalLocation("target/A20161224.1030-1045.bin.gz")
+ .compression("gzip")
+ .fileFormatType("org.3GPP.32.435#measCollec")
+ .fileFormatVersion("V10")
+ .build();
+
private static final String EXPECTED_RESULT =
- "{\"name\":\"A20161224.1030-1045.bin.gz\",\"location\":\"target/A20161224.1030-1045.bin.gz\","
- + "\"compression\":\"gzip\",\"fileFormatType\":\"org.3GPP.32.435#measCollec\","
- + "\"fileFormatVersion\":\"V10\"}";
-
+ "{\"productName\":\"NrRadio\","
+ + "\"vendorName\":\"Ericsson\","
+ + "\"lastEpochMicrosec\":\"8745745764578\","
+ + "\"sourceName\":\"oteNB5309\","
+ + "\"startEpochMicrosec\":\"8745745764578\","
+ + "\"timeZoneOffset\":\"UTC+05:00\","
+ + "\"name\":\"A20161224.1030-1045.bin.gz\","
+ + "\"location\":\"ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1145.bin.gz\","
+ + "\"internalLocation\":\"target/A20161224.1030-1045.bin.gz\","
+ + "\"compression\":\"gzip\","
+ + "\"fileFormatType\":\"org.3GPP.32.435#measCollec\","
+ + "\"fileFormatVersion\":\"V10\"}";
+ // @formatter:on
@Test
void createJsonBody_shouldReturnJsonInString() {
assertEquals(EXPECTED_RESULT, CommonFunctions.createJsonBody(model));
import org.junit.jupiter.api.Test;
public class ConsumerDmaapModelTest {
-
+ private static final String PRODUCT_NAME = "NrRadio";
+ private static final String VENDOR_NAME = "Ericsson";
+ private static final String LAST_EPOCH_MICROSEC = "8745745764578";
+ private static final String SOURCE_NAME = "oteNB5309";
+ private static final String START_EPOCH_MICROSEC = "8745745764578";
+ private static final String TIME_ZONE_OFFSET = "UTC+05:00";
private static final String NAME = "A20161224.1030-1045.bin.gz";
- private static final String LOCATION = "target/A20161224.1030-1045.bin.gz";
+ private static final String LOCATION = "ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1145.bin.gz";
+ private static final String INTERNAL_LOCATION = "target/A20161224.1030-1045.bin.gz";
private static final String COMPRESSION = "gzip";
private static final String FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
private static final String FILE_FORMAT_VERSION = "V10";
@Test
public void consumerDmaapModelBuilder_shouldBuildAnObject() {
-
- ConsumerDmaapModel consumerDmaapModel =
- ImmutableConsumerDmaapModel.builder().name(NAME).location(LOCATION).compression(COMPRESSION)
- .fileFormatType(FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+ // @formatter:off
+ ConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder()
+ .productName(PRODUCT_NAME)
+ .vendorName(VENDOR_NAME)
+ .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
+ .sourceName(SOURCE_NAME)
+ .startEpochMicrosec(START_EPOCH_MICROSEC)
+ .timeZoneOffset(TIME_ZONE_OFFSET)
+ .name(NAME)
+ .location(LOCATION)
+ .internalLocation(INTERNAL_LOCATION)
+ .compression(COMPRESSION)
+ .fileFormatType(FILE_FORMAT_TYPE)
+ .fileFormatVersion(FILE_FORMAT_VERSION)
+ .build();
+ // @formatter:on
Assertions.assertNotNull(consumerDmaapModel);
+ Assertions.assertEquals(PRODUCT_NAME, consumerDmaapModel.getProductName());
+ Assertions.assertEquals(VENDOR_NAME, consumerDmaapModel.getVendorName());
+ Assertions.assertEquals(LAST_EPOCH_MICROSEC, consumerDmaapModel.getLastEpochMicrosec());
+ Assertions.assertEquals(SOURCE_NAME, consumerDmaapModel.getSourceName());
+ Assertions.assertEquals(START_EPOCH_MICROSEC, consumerDmaapModel.getStartEpochMicrosec());
+ Assertions.assertEquals(TIME_ZONE_OFFSET, consumerDmaapModel.getTimeZoneOffset());
Assertions.assertEquals(NAME, consumerDmaapModel.getName());
Assertions.assertEquals(LOCATION, consumerDmaapModel.getLocation());
+ Assertions.assertEquals(INTERNAL_LOCATION, consumerDmaapModel.getInternalLocation());
Assertions.assertEquals(COMPRESSION, consumerDmaapModel.getCompression());
Assertions.assertEquals(FILE_FORMAT_TYPE, consumerDmaapModel.getFileFormatType());
Assertions.assertEquals(FILE_FORMAT_VERSION, consumerDmaapModel.getFileFormatVersion());
private static final String X_ATT_DR_META = "X-ATT-DR-META";
private static final String NAME_JSON_TAG = "name";
- private static final String LOCATION_JSON_TAG = "location";
+ private static final String INTERNAL_LOCATION_JSON_TAG = "internalLocation";
private static final String URI_SEPARATOR = "/";
private static final String DEFAULT_FEED_ID = "1";
addUserCredentialsToHead(headers);
IFileSystemResource fileSystemResource = getFileSystemResource();
- fileSystemResource.setPath(consumerDmaapModel.getLocation());
+ fileSystemResource.setPath(consumerDmaapModel.getInternalLocation());
InputStream fileInputStream = fileSystemResource.getInputStream();
HttpEntity<byte[]> request = addFileToRequest(fileInputStream, headers);
private void addMetaDataToHead(ConsumerDmaapModel consumerDmaapModel, HttpHeaders headers) {
JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(consumerDmaapModel));
metaData.getAsJsonObject().remove(NAME_JSON_TAG).getAsString();
- metaData.getAsJsonObject().remove(LOCATION_JSON_TAG);
+ metaData.getAsJsonObject().remove(INTERNAL_LOCATION_JSON_TAG);
headers.set(X_ATT_DR_META, metaData.toString());
}
private HttpEntity<byte[]> addFileToRequest(InputStream inputStream, HttpHeaders headers)
class DmaapProducerReactiveHttpClientTest {
private static final String FILE_NAME = "A20161224.1030-1045.bin.gz";
- private static final String LOCATION_JSON_TAG = "location";
+ private static final String INTERNAL_LOCATION_JSON_TAG = "internalLocation";
private static final String NAME_JSON_TAG = "name";
private static final String X_ATT_DR_META = "X-ATT-DR-META";
when(dmaapPublisherConfigurationMock.dmaapContentType()).thenReturn(APPLICATION_OCTET_STREAM_CONTENT_TYPE);
when(dmaapPublisherConfigurationMock.dmaapTopicName()).thenReturn(PUBLISH_TOPIC);
- consumerDmaapModel = ImmutableConsumerDmaapModel.builder().name(FILE_NAME)
- .location("target/A20161224.1030-1045.bin.gz").compression("gzip")
- .fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
+ // @formatter:off
+ consumerDmaapModel = ImmutableConsumerDmaapModel.builder()
+ .productName("NrRadio")
+ .vendorName("Ericsson")
+ .lastEpochMicrosec("8745745764578")
+ .sourceName("oteNB5309")
+ .startEpochMicrosec("8745745764578")
+ .timeZoneOffset("UTC+05:00")
+ .name("A20161224.1030-1045.bin.gz")
+ .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1145.bin.gz")
+ .internalLocation("target/A20161224.1030-1045.bin.gz")
+ .compression("gzip")
+ .fileFormatType("org.3GPP.32.435#measCollec")
+ .fileFormatVersion("V10")
+ .build();
+ //formatter:on
dmaapProducerReactiveHttpClient = new DmaapProducerReactiveHttpClient(dmaapPublisherConfigurationMock);
dmaapProducerReactiveHttpClient.setFileSystemResource(fileSystemResourceMock);
JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(consumerDmaapModel));
metaData.getAsJsonObject().remove(NAME_JSON_TAG).getAsString();
- metaData.getAsJsonObject().remove(LOCATION_JSON_TAG);
+ metaData.getAsJsonObject().remove(INTERNAL_LOCATION_JSON_TAG);
headers.set(X_ATT_DR_META, metaData.toString());
String plainCreds = "dradmin" + ":" + "dradmin";
Code Review / dcaegen2 / collectors / datafile.git / commitdiff