Skip to content

Commit

Permalink
writer: add provenance info to each call from dc-interfaces
Browse files Browse the repository at this point in the history
  • Loading branch information
Peter Moser committed Mar 2, 2022
1 parent 8ea54bb commit fe62b05
Show file tree
Hide file tree
Showing 4 changed files with 151 additions and 70 deletions.
2 changes: 1 addition & 1 deletion calls.http
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ GET {{host}}/json/getDateOfLastRecord/MeteoStation?stationId=T0009&period=900
Authorization: Bearer {{authtoken}}

### Post some data to the writer
POST {{host}}/json/provenance
POST {{host}}/json/provenance?prn=test&prv=11111
Content-Type: application/json
Authorization: Bearer {{authtoken}}

Expand Down
113 changes: 67 additions & 46 deletions dc-interface/src/main/java/it/bz/idm/bdp/json/JSONPusher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,12 +27,16 @@

import javax.annotation.PostConstruct;

import org.springframework.http.HttpEntity;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import it.bz.idm.bdp.DataPusher;
import it.bz.idm.bdp.dto.DataMapDto;
import it.bz.idm.bdp.dto.DataTypeDto;
import it.bz.idm.bdp.dto.ProvenanceDto;
import it.bz.idm.bdp.dto.RecordDtoImpl;
import it.bz.idm.bdp.dto.StationDto;
import it.bz.idm.bdp.dto.StationList;
Expand Down Expand Up @@ -68,25 +72,36 @@ public void init() {
public Object pushData(String datasourceName, DataMapDto<? extends RecordDtoImpl> dto) {
this.pushProvenance();
dto.setProvenance(this.provenance.getUuid());
return restTemplate.postForObject(
url + PUSH_RECORDS + "{datasourceName}?prn={}&prv={}",
dto,
Object.class,
datasourceName,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
return restTemplate
.exchange(
url + PUSH_RECORDS + "{datasourceName}?prn={}&prv={}",
HttpMethod.POST,
new HttpEntity<DataMapDto<? extends RecordDtoImpl>>(dto),
Object.class,
datasourceName,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
)
.getBody();
}

private void pushProvenance() {
String provenanceUuid = restTemplate.postForObject(
// We know that the provenance exist, and which UUID it has.
// So we do not need to get that information again from the DB
// This approach assumes, that the DB will not change from any
// other caller.
if (this.provenance.getUuid() != null) {
return;
}
ResponseEntity<String> provenanceUuid = restTemplate.exchange(
url + PROVENANCE + "?prn={}&prv={}",
this.provenance,
HttpMethod.POST,
new HttpEntity<ProvenanceDto>(this.provenance),
String.class,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
this.provenance.setUuid(provenanceUuid);
this.provenance.setUuid(provenanceUuid.getBody());
}

public Object pushData(DataMapDto<? extends RecordDtoImpl> dto) {
Expand All @@ -102,27 +117,33 @@ public Object syncStations(StationList data) {
public Object syncStations(String datasourceName, StationList data) {
if (data == null)
return null;
return restTemplate.postForObject(
url + SYNC_STATIONS + "{datasourceName}?prn={}&prv={}",
data,
Object.class,
datasourceName,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
return restTemplate
.exchange(
url + SYNC_STATIONS + "{datasourceName}?prn={}&prv={}",
HttpMethod.POST,
new HttpEntity<StationList>(data),
Object.class,
datasourceName,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
)
.getBody();
}

@Override
public Object syncDataTypes(String datasourceName, List<DataTypeDto> data) {
if (data == null)
return null;
return restTemplate.postForObject(
url + SYNC_DATA_TYPES + "?prn={}&prv={}",
data,
Object.class,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
return restTemplate
.exchange(
url + SYNC_DATA_TYPES + "?prn={}&prv={}",
HttpMethod.POST,
new HttpEntity<List<DataTypeDto>>(data),
Object.class,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
)
.getBody();
}

public Object syncDataTypes(List<DataTypeDto> data) {
Expand All @@ -131,16 +152,17 @@ public Object syncDataTypes(List<DataTypeDto> data) {

@Override
public Object getDateOfLastRecord(String stationCode, String dataType, Integer period) {
return restTemplate.getForObject(
url + GET_DATE_OF_LAST_RECORD + "{datasourceName}/?stationId={stationId}&typeId={dataType}&period={period}&prn={}&prv={}",
Date.class,
this.integreenTypology,
stationCode,
dataType,
period,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
return restTemplate
.getForObject(
url + GET_DATE_OF_LAST_RECORD + "{datasourceName}/?stationId={stationId}&typeId={dataType}&period={period}&prn={}&prv={}",
Date.class,
this.integreenTypology,
stationCode,
dataType,
period,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
}

@Override
Expand All @@ -150,16 +172,15 @@ public void connectToDataCenterCollector() {

@Override
public List<StationDto> fetchStations(String datasourceName, String origin) {
if (datasourceName == null)
datasourceName = this.integreenTypology;
StationDto[] object = restTemplate.getForObject(
url + STATIONS +"{datasourceName}/?origin={origin}&prn={}&prv={}",
StationDto[].class,
datasourceName,
origin,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
StationDto[] object = restTemplate
.getForObject(
url + STATIONS +"{datasourceName}/?origin={origin}&prn={}&prv={}",
StationDto[].class,
datasourceName == null ? this.integreenTypology : datasourceName,
origin,
provenance.getDataCollector(),
provenance.getDataCollectorVersion()
);
return Arrays.asList(object);
}

Expand Down
100 changes: 80 additions & 20 deletions dc-interface/src/main/java/it/bz/idm/bdp/json/NonBlockingJSONPusher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,13 +72,33 @@ public void init() {
public Object pushData(String datasourceName, DataMapDto<? extends RecordDtoImpl> dto) {
this.pushProvenance();
dto.setProvenance(this.provenance.getUuid());
return client.post().uri(PUSH_RECORDS + datasourceName).body(Mono.just(dto), Object.class).retrieve()
.bodyToMono(Object.class).block();
return client
.post()
.uri(uriBuilder -> uriBuilder
.path(PUSH_RECORDS + datasourceName)
.queryParam("prn", provenance.getDataCollector())
.queryParam("prv", provenance.getDataCollectorVersion())
.build()
)
.body(Mono.just(dto), Object.class)
.retrieve()
.bodyToMono(Object.class)
.block();
}

private void pushProvenance() {
String provenanceUuid = client.post().uri(PROVENANCE).body(Mono.just(this.provenance), ProvenanceDto.class).retrieve()
.bodyToMono(String.class).block();
String provenanceUuid = client
.post()
.uri(uriBuilder -> uriBuilder
.path(PROVENANCE)
.queryParam("prn", provenance.getDataCollector())
.queryParam("prv", provenance.getDataCollectorVersion())
.build()
)
.body(Mono.just(this.provenance), ProvenanceDto.class)
.retrieve()
.bodyToMono(String.class)
.block();
this.provenance.setUuid(provenanceUuid);
}

Expand All @@ -95,16 +115,36 @@ public Object syncStations(StationList data) {
public Object syncStations(String datasourceName, StationList data) {
if (data == null)
return null;
return client.post().uri(SYNC_STATIONS + datasourceName).body(Mono.just(data), Object.class).retrieve()
.bodyToMono(Object.class).block();
return client
.post()
.uri(uriBuilder -> uriBuilder
.path(SYNC_STATIONS + datasourceName)
.queryParam("prn", provenance.getDataCollector())
.queryParam("prv", provenance.getDataCollectorVersion())
.build()
)
.body(Mono.just(data), Object.class)
.retrieve()
.bodyToMono(Object.class)
.block();
}

@Override
public Object syncDataTypes(String datasourceName, List<DataTypeDto> data) {
if (data == null)
return null;
return client.post().uri(SYNC_DATA_TYPES).body(Mono.just(data), Object.class).retrieve()
.bodyToMono(Object.class).block();
return client
.post()
.uri(uriBuilder -> uriBuilder
.path(SYNC_DATA_TYPES)
.queryParam("prn", provenance.getDataCollector())
.queryParam("prv", provenance.getDataCollectorVersion())
.build()
)
.body(Mono.just(data), Object.class)
.retrieve()
.bodyToMono(Object.class)
.block();
}

public Object syncDataTypes(List<DataTypeDto> data) {
Expand All @@ -113,12 +153,20 @@ public Object syncDataTypes(List<DataTypeDto> data) {

@Override
public Object getDateOfLastRecord(String stationCode, String dataType, Integer period) {
return client.get().uri(uriBuilder -> uriBuilder
.path(GET_DATE_OF_LAST_RECORD + this.integreenTypology)
.queryParam("stationId", stationCode)
.queryParam("typeId", dataType)
.queryParam("period", period).build())
.retrieve().bodyToMono(Date.class).block();
return client
.get()
.uri(uriBuilder -> uriBuilder
.path(GET_DATE_OF_LAST_RECORD + this.integreenTypology)
.queryParam("stationId", stationCode)
.queryParam("typeId", dataType)
.queryParam("period", period)
.queryParam("prn", provenance.getDataCollector())
.queryParam("prv", provenance.getDataCollectorVersion())
.build()
)
.retrieve()
.bodyToMono(Date.class)
.block();
}

@Override
Expand All @@ -128,11 +176,18 @@ public void connectToDataCenterCollector() {

@Override
public List<StationDto> fetchStations(String datasourceName, String origin) {
final String uri = STATIONS + datasourceName;
if (datasourceName == null)
datasourceName = this.integreenTypology;
StationDto[] object = client.get().uri(uriBuilder->uriBuilder.path(uri).queryParam("origin", "{origin}").build(origin)).retrieve()
.bodyToMono(StationDto[].class).block();
StationDto[] object = client
.get()
.uri(uriBuilder->uriBuilder
.path(datasourceName == null ? STATIONS + this.integreenTypology : STATIONS + datasourceName)
.queryParam("origin", "{origin}")
.queryParam("prn", provenance.getDataCollector())
.queryParam("prv", provenance.getDataCollectorVersion())
.build(origin)
)
.retrieve()
.bodyToMono(StationDto[].class)
.block();
return Arrays.asList(object);
}

Expand All @@ -147,7 +202,12 @@ public Object addEvents(List<EventDto> dtos) {
}
return client
.post()
.uri(EVENTS)
.uri(uriBuilder->uriBuilder
.path(EVENTS)
.queryParam("prn", provenance.getDataCollector())
.queryParam("prv", provenance.getDataCollectorVersion())
.build()
)
.body(Mono.just(dtos), Object.class)
.retrieve()
.bodyToMono(Object.class)
Expand Down
6 changes: 3 additions & 3 deletions writer/src/main/java/it/bz/idm/bdp/writer/JsonController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -100,8 +100,8 @@ public ResponseEntity<Date> dateOfLastRecord(
HttpServletRequest request,
@PathVariable("integreenTypology") String stationType,
@RequestParam("stationId") String stationId,
@RequestParam(value="typeId", required=false) String typeId,
@RequestParam(value="period", required=false) Integer period,
@RequestParam(value = "typeId", required = false) String typeId,
@RequestParam(value = "period", required = false) Integer period,
@RequestParam(value = "prn", required = false) String proveanceName,
@RequestParam(value = "prv", required = false) String provenanceVersion
) {
Expand All @@ -113,7 +113,7 @@ public ResponseEntity<Date> dateOfLastRecord(
public Object stationsGetList(
HttpServletRequest request,
@PathVariable("integreenTypology") String stationType,
@RequestParam(value="origin", required=false) String origin,
@RequestParam(value = "origin", required = false) String origin,
@RequestParam(value = "prn", required = false) String proveanceName,
@RequestParam(value = "prv", required = false) String provenanceVersion
) {
Expand Down

0 comments on commit fe62b05

Please sign in to comment.