Skip to content

Commit afd5da2

Browse files
valentin-gauthier-geosirisvalentin-gauthier-geosiris
committed
bugfix
1 parent ec438f2 commit afd5da2

5 files changed

Lines changed: 71 additions & 158 deletions

File tree

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
<groupId>com.geosiris</groupId>
55
<artifactId>etpproto-java</artifactId>
66
<name>${project.groupId}:${project.artifactId}</name>
7-
<version>1.0.2</version>
7+
<version>1.0.3</version>
88
<organization>
99
<name>Geosiris</name>
1010
<url>http://www.geosiris.com</url>

‎src/example/com/geosiris/etp/example/Main.java‎

Lines changed: 23 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,9 @@
1515
*/
1616
package com.geosiris.etp.example;
1717

18-
import Energistics.Etp.v12.Datatypes.AnyArray;
19-
import Energistics.Etp.v12.Datatypes.ArrayOfDouble;
18+
import Energistics.Etp.v12.Datatypes.*;
2019
import Energistics.Etp.v12.Datatypes.DataArrayTypes.DataArray;
21-
import Energistics.Etp.v12.Datatypes.DataValue;
2220
import Energistics.Etp.v12.Datatypes.Object.ContextScopeKind;
23-
import Energistics.Etp.v12.Datatypes.ServerCapabilities;
2421
import Energistics.Etp.v12.Protocol.DataArray.PutDataArrays;
2522
import Energistics.Etp.v12.Protocol.DataArray.PutDataArraysResponse;
2623
import com.geosiris.etp.communication.ClientInfo;
@@ -61,10 +58,10 @@ public static void main(String[] args) throws Exception {
6158
// etpClientTest2(args);
6259
// test_big_message(args);
6360
// test_multiple_data_array(args);
64-
// test_get_big_data_array(args);
61+
test_get_big_data_array(args);
6562
// test_multiple_data_arraySmall(args);
6663
// test_multiple_data_REST(args);
67-
test_get_dataspaces(args);
64+
// test_get_dataspaces(args);
6865
// test_get_resources(args);
6966
// test_big_messageREST(args);
7067
}
@@ -217,12 +214,13 @@ public static void test_multiple_data_array(String[] args) throws Exception {
217214
}
218215

219216
public static void test_get_big_data_array(String[] args) throws Exception {
220-
ETPClient client = getClient(1<<16, args);
217+
ETPClient client = getClient(1<<22, args);
221218

222219
String uri = "eml:///dataspace('volve-eqn-plus')/resqml20.obj_Grid2dRepresentation(3a45fb70-8ba9-4341-a701-0f514270ba9c)";
223220
try {
224-
ETPHelper.sendGetDataArray(client, uri, "/RESQML/3a45fb70-8ba9-4341-a701-0f514270ba9c/points_patch0", 50000);
225-
Thread.sleep(3000000);
221+
// ETPHelper.sendGetDataArray(client, uri, "/RESQML/3a45fb70-8ba9-4341-a701-0f514270ba9c/points_patch0", 50000);
222+
logger.info(ETPHelper.getMultipleDataArrays(client, uri, List.of("/RESQML/3a45fb70-8ba9-4341-a701-0f514270ba9c/points_patch0"), 50000));
223+
// Thread.sleep(3000000);
226224
}finally {
227225
client.closeClient();
228226
}
@@ -385,12 +383,22 @@ public static void etpClientTest2(String args[]) {
385383
protocolHandlers.put(CoreHandler_DefaultPrinter.protocol, new CoreHandler_DefaultPrinter());
386384
protocolHandlers.put(DiscoveryHandler_DefaultPrinter.protocol, new DiscoveryHandler_DefaultPrinter());
387385

388-
ServerCapabilities caps = ServerCapabilities.newBuilder()
389-
.setApplicationVersion("1.0.2")
390-
.setApplicationName("GeosirisIlabTest")
391-
.build();
386+
// ServerCapabilities caps = ServerCapabilities.newBuilder()
387+
// .setApplicationVersion("1.0.2")
388+
// .setApplicationName("GeosirisIlabTest")
389+
// .setContactInformation(Contact.newBuilder()
390+
// .setContactPhone("")
391+
// .setContactName("Valentin")
392+
// .setContactEmail("valentin.gauthier@geosiris.com")
393+
// .setOrganizationName("Geosiris")
394+
// .build())
395+
// .setSupportedDataObjects(new ArrayList<>())
396+
// .setSupportedProtocols(new ArrayList<>(
397+
//
398+
// ))
399+
// .build();
392400

393-
ETPConnection etpConnection = new ETPConnection(ConnectionType.CLIENT, caps, clientInfo, protocolHandlers);
401+
ETPConnection etpConnection = new ETPConnection(ConnectionType.CLIENT, new ServerCapabilities(), clientInfo, protocolHandlers);
394402

395403
ETPClient etpClient = null;
396404
if(true||args.length >= 3) {
@@ -409,44 +417,7 @@ public static void etpClientTest2(String args[]) {
409417
}
410418

411419
public static void etpClientTest(String args[]){
412-
logger.info("Usage : java -jar myfile.jar [SERVER_URL] [LOGIN] [PASSWORD]");
413-
414-
HttpURI etpServerUri = null;
415-
String login = "";
416-
String password = "";
417-
418-
if (args.length > 0) {
419-
try { etpServerUri = new HttpURI(args[0]);
420-
}catch(Exception e) {e.printStackTrace();}
421-
}
422-
if (args.length > 1) {
423-
try { login = args[1];
424-
}catch(Exception e) {e.printStackTrace();}
425-
}
426-
if (args.length > 2) {
427-
try { password = args[2];
428-
}catch(Exception e) {e.printStackTrace();}
429-
}
430-
logger.info(etpServerUri);
431-
432-
ClientInfo clientInfo = new ClientInfo(etpServerUri);
433-
Map<CommunicationProtocol, ProtocolHandler> protocolHandlers = new HashMap<>();
434-
protocolHandlers.put(CoreHandler_DefaultPrinter.protocol, new CoreHandler_DefaultPrinter());
435-
protocolHandlers.put(StoreHandler_DefaultPrinter.protocol, new StoreHandler_DefaultPrinter());
436-
protocolHandlers.put(DataspaceHandler_DefaultPrinter.protocol, new DataspaceHandler_DefaultPrinter());
437-
protocolHandlers.put(DataArrayHandler_DefaultPrinter.protocol, new DataArrayHandler_DefaultPrinter());
438-
protocolHandlers.put(DiscoveryHandler_DefaultPrinter.protocol, new DiscoveryHandler_DefaultPrinter());
439-
440-
ETPConnection etpConnection = new ETPConnection(ConnectionType.CLIENT, new ServerCapabilities(), clientInfo, protocolHandlers);
441-
442-
ETPClient etpClient = null;
443-
if(true||args.length >= 3) {
444-
etpClient = ETPClient.getInstanceWithAuth_Basic(etpServerUri, etpConnection, 2000, login, password);
445-
}else if(args.length >= 3) {
446-
etpClient = ETPClient.getInstanceWithAuth_Token(etpServerUri, etpConnection, 10000, args[1]);
447-
}
448-
449-
420+
ETPClient etpClient = getClient(args);
450421

451422
if(etpClient != null){
452423
{

‎src/main/java/com/geosiris/etp/utils/ETPHelper.java‎

Lines changed: 45 additions & 103 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
import Energistics.Etp.v12.Datatypes.AnyArray;
1919
import Energistics.Etp.v12.Datatypes.AnyLogicalArrayType;
20-
import Energistics.Etp.v12.Datatypes.ArrayOfFloat;
2120
import Energistics.Etp.v12.Datatypes.DataArrayTypes.*;
2221
import Energistics.Etp.v12.Datatypes.Object.*;
2322
import Energistics.Etp.v12.Protocol.DataArray.*;
@@ -28,19 +27,15 @@
2827
import Energistics.Etp.v12.Protocol.Store.PutDataObjects;
2928
import com.geosiris.etp.communication.Message;
3029
import com.geosiris.etp.websocket.ETPClient;
31-
import org.apache.avro.reflect.MapEntry;
32-
import org.apache.avro.specific.SpecificRecordBase;
3330
import org.apache.logging.log4j.LogManager;
3431
import org.apache.logging.log4j.Logger;
3532
import org.w3c.dom.Document;
3633

3734
import java.io.File;
3835
import java.io.FileNotFoundException;
39-
import java.lang.reflect.InvocationTargetException;
4036
import java.nio.charset.StandardCharsets;
4137
import java.util.*;
4238
import java.util.stream.Collectors;
43-
import java.util.stream.IntStream;
4439

4540
public class ETPHelper {
4641
public static final Logger logger = LogManager.getLogger(ETPHelper.class);
@@ -338,7 +333,6 @@ public static Map<String, List<Number>> getMultipleDataArrays(ETPClient client,
338333
logger.debug("\t[{}]", String.join(",\n", subArrays.keySet()));
339334

340335
assert subArrays.size() + fullArrays.stream().map(x -> x.r().size()).reduce(0, (a,b) -> a + b) == datasets_paths.size();
341-
// if(true) return result;
342336

343337
List<Long> msgIdsToWait = new ArrayList<>();
344338
for(Pair<Long, List<String>> fk: fullArrays){
@@ -360,30 +354,25 @@ public static Map<String, List<Number>> getMultipleDataArrays(ETPClient client,
360354
Map<Long, String> msgIdsToWaitSubArray = new HashMap<>();
361355
for(Map.Entry<String, DataArrayMetadata> e_subArray: subArrays.entrySet()){
362356
List<Long> dimensions = e_subArray.getValue().getDimensions();
363-
long thisSize = e_subArray.getValue().getDimensions().stream().reduce(1L, (a, b) -> a * b) * eltSize;
364357
Map<CharSequence, GetDataSubarraysType> mapSubArr = new HashMap<>();
365358

366-
long nbSplit = thisSize * eltSize / maxMsgSize;
367-
for(int i=0; i<=nbSplit; i++) {
368-
long sub_start = ((i) * maxMsgSize / eltSize);
369-
long sub_stop = Math.min((i + 1) * maxMsgSize / eltSize, thisSize / eltSize);
370-
long sub_size = sub_stop - sub_start;
359+
long startLineIdx = 0;
360+
long endLineIdx = 0;
361+
long sizeOfLine = dimensions.size() > 1 ? dimensions.subList(1, dimensions.size()).stream().reduce(1L, (a, b) -> a * b) : 1;
362+
long nbLinePerMsg = (maxMsgSize - securitySize) / (sizeOfLine * eltSize);
363+
while(endLineIdx < dimensions.get(0)){
364+
endLineIdx = Math.min(startLineIdx + nbLinePerMsg, dimensions.get(0));
371365
List<Long> counts = new ArrayList<>();
372366
List<Long> starts = new ArrayList<>();
373-
List<Long> dim_reversed = new ArrayList<>(dimensions);
374-
Collections.reverse(dim_reversed);
375-
long tmp_sub_start = sub_start;
376-
for(int di=0; di < dimensions.size(); di++){
377-
if(di < dimensions.size() - 1) {
378-
Long lineSize = dimensions.subList(di + 1, dimensions.size()).stream().reduce(1L, (subtotal, element) -> subtotal * element);
379-
starts.add(tmp_sub_start / lineSize);
380-
tmp_sub_start -= starts.get(di) * lineSize;
381-
counts.add(sub_size/lineSize);
382-
}else {
383-
starts.add(tmp_sub_start);
384-
counts.add(Math.min(dimensions.get(di), sub_size));
385-
}
367+
368+
counts.add(endLineIdx - startLineIdx);
369+
starts.add(startLineIdx);
370+
371+
for(int i=1; i<dimensions.size(); i++){
372+
counts.add(dimensions.get(i));
373+
starts.add(0L);
386374
}
375+
logger.debug("SubArr : Counts {} : Starts {}", counts , starts);
387376
mapSubArr.put(String.valueOf(mapSubArr.size()),
388377
GetDataSubarraysType.newBuilder()
389378
.setCounts(counts)
@@ -394,18 +383,20 @@ public static Map<String, List<Number>> getMultipleDataArrays(ETPClient client,
394383
.build()
395384
).build()
396385
);
397-
}
398-
GetDataSubarrays gdsa = GetDataSubarrays.newBuilder()
386+
GetDataSubarrays gdsa = GetDataSubarrays.newBuilder()
399387
.setDataSubarrays(mapSubArr).build();
400-
long msgId = client.send(gdsa);
401-
msgIdsToWaitSubArray.put(msgId, e_subArray.getKey());
388+
long msgId = client.send(gdsa);
389+
msgIdsToWaitSubArray.put(msgId, e_subArray.getKey());
390+
391+
startLineIdx = startLineIdx + nbLinePerMsg;
392+
}
402393
}
403394

404-
Map<String, List<GetDataSubarraysResponse>> subArrayResps = new HashMap<>();
395+
Map<String, List<Message>> subArrayResps = new HashMap<>();
405396

406397
for(Map.Entry<Long, List<Message>> e_answers: client.getEtpClientSession().waitForResponse(msgIdsToWait, 500000).entrySet()){
407398
List<Message> answers = e_answers.getValue();
408-
for(Message answer: answers.stream().sorted((m0, m1) -> Integer.compare((int) m0.getHeader().getMessageId(), (int) m1.getHeader().getMessageId())).collect(Collectors.toList())) {
399+
for(Message answer: answers.stream().sorted(Comparator.comparingInt(m0 -> (int) m0.getHeader().getMessageId())).collect(Collectors.toList())) {
409400
if (answer.getBody() instanceof GetDataArraysResponse) {
410401
GetDataArraysResponse gdar = ((GetDataArraysResponse) answer.getBody());
411402
for(Map.Entry<CharSequence, DataArray> e_gdar: gdar.getDataArrays().entrySet()){
@@ -417,15 +408,6 @@ public static Map<String, List<Number>> getMultipleDataArrays(ETPClient client,
417408
}
418409

419410
}
420-
// } else if (answer.getBody() instanceof GetDataSubarraysResponse) {
421-
// String identifier = msgIdsToWaitSubArray.get(e_answers.getKey());
422-
// logger.debug("identifier : {}", identifier);
423-
// logger.debug(String.join(", \n", ((GetDataSubarraysResponse) answer.getBody()).getDataSubarrays().keySet()));
424-
// if(!subArrayResps.containsKey(identifier)){
425-
// subArrayResps.put(identifier, new ArrayList<>());
426-
// }
427-
// subArrayResps.get(identifier).add(((GetDataSubarraysResponse) answer.getBody()));
428-
429411
} else {
430412
logger.debug("Unexpected answer for msg : {}", answer.getHeader().getCorrelationId());
431413
}
@@ -434,15 +416,16 @@ public static Map<String, List<Number>> getMultipleDataArrays(ETPClient client,
434416

435417
for(Map.Entry<Long, List<Message>> e_answers: client.getEtpClientSession().waitForResponse(new ArrayList<>(msgIdsToWaitSubArray.keySet()), 500000).entrySet()){
436418
List<Message> answers = e_answers.getValue();
437-
for(Message answer: answers.stream().sorted((m0, m1) -> Integer.compare((int) m0.getHeader().getMessageId(), (int) m1.getHeader().getMessageId())).collect(Collectors.toList())) {
419+
for(Message answer: answers.stream().sorted(Comparator.comparingInt(m0 -> (int) m0.getHeader().getMessageId())).collect(Collectors.toList())) {
438420
if (answer.getBody() instanceof GetDataSubarraysResponse) {
439421
String identifier = msgIdsToWaitSubArray.get(e_answers.getKey());
440422
logger.debug("identifier : {}", identifier);
441423
logger.debug(String.join(", \n", ((GetDataSubarraysResponse) answer.getBody()).getDataSubarrays().keySet()));
442424
if(!subArrayResps.containsKey(identifier)){
443425
subArrayResps.put(identifier, new ArrayList<>());
444426
}
445-
subArrayResps.get(identifier).add(((GetDataSubarraysResponse) answer.getBody()));
427+
// subArrayResps.get(identifier).add(((GetDataSubarraysResponse) answer.getBody()));
428+
subArrayResps.get(identifier).add(answer);
446429

447430
} else {
448431
logger.debug("Unexpected answer for msg : {}", answer.getHeader().getCorrelationId());
@@ -451,9 +434,11 @@ public static Map<String, List<Number>> getMultipleDataArrays(ETPClient client,
451434
}
452435

453436
// For subArraysResponse : sort and reduce results for each array
454-
for(Map.Entry<String, List<GetDataSubarraysResponse>> sub_arr_r: subArrayResps.entrySet()){
455-
Map<CharSequence, DataArray> allSubArrayParts = sub_arr_r.getValue().stream().map(gdsr -> gdsr.getDataSubarrays().entrySet()).flatMap(Set::stream)
456-
.collect(Collectors.toMap(Map.Entry<CharSequence, DataArray>::getKey, Map.Entry<CharSequence, DataArray>::getValue));
437+
for(Map.Entry<String, List<Message>> sub_arr_r: subArrayResps.entrySet()){
438+
/*Map<CharSequence, DataArray> allSubArrayParts =
439+
sub_arr_r.getValue().stream()
440+
.map(m_gdsr -> ((GetDataSubarraysResponse)m_gdsr.getBody()).getDataSubarrays().entrySet()).flatMap(Set::stream)
441+
.collect(Collectors.toMap(Map.Entry<CharSequence, DataArray>::getKey, Map.Entry<CharSequence, DataArray>::getValue));
457442
List<Number> numbers = allSubArrayParts.entrySet().stream()
458443
.sorted(Comparator.comparingInt(e -> Integer.getInteger(e.getKey().toString())))
459444
.map(e -> {
@@ -464,67 +449,24 @@ public static Map<String, List<Number>> getMultipleDataArrays(ETPClient client,
464449
}
465450
})
466451
.flatMap(List::stream).collect(Collectors.toList());
467-
result.put(sub_arr_r.getKey(),numbers);
452+
result.put(sub_arr_r.getKey(),numbers);*/
453+
List<Number> das = sub_arr_r.getValue().stream()
454+
.sorted(Comparator.comparingLong(m0 -> m0.getHeader().getMessageId()))
455+
.map(m -> ((GetDataSubarraysResponse) m.getBody()).getDataSubarrays().values())
456+
.flatMap(Collection::stream)
457+
.map(e -> {
458+
try {
459+
return (List<Number>) ETPUtils.getAttributeValue(e.getData().getItem(), "values");
460+
} catch (Exception ex) {
461+
throw new RuntimeException(ex);
462+
}
463+
}).filter(Objects::nonNull)
464+
.flatMap(List::stream)
465+
.collect(Collectors.toList());
466+
result.put(sub_arr_r.getKey(), das);
468467
}
469468

470469
return result;
471-
472-
473-
/*Map<CharSequence, DataArrayIdentifier> mapIdentifier = new HashMap<>();
474-
for(String ds_path: datasets_paths){
475-
DataArrayIdentifier identifier = DataArrayIdentifier.newBuilder()
476-
.setUri(uri)
477-
.setPathInResource(ds_path)
478-
.build();
479-
mapIdentifier.put(String.valueOf(mapIdentifier.size()), identifier);
480-
}
481-
482-
GetDataArrays gda = GetDataArrays.newBuilder()
483-
.setDataArrays(mapIdentifier).build();
484-
long msg_id = client.send(gda);
485-
logger.debug(msg_id + ") GetDataSubArray sent " + gda);
486-
List<Message> resp = client.getEtpClientSession().waitForResponse(msg_id, timeoutMS);
487-
System.out.println("End Wait");
488-
// if(true) return null;
489-
Map<CharSequence, DataArray> da_entries = new HashMap<>();
490-
for(Message m: resp){
491-
if(m.getBody() instanceof GetDataArraysResponse){
492-
GetDataArraysResponse gdar = (GetDataArraysResponse) m.getBody();
493-
da_entries.putAll(gdar.getDataArrays());
494-
// for(Map.Entry<CharSequence, DataArray> entry: gdar.getDataArrays().entrySet()){
495-
// try {
496-
// da_entries.set(Integer.getInteger(String.valueOf(entry.getKey())), entry.getValue());
497-
// }catch (Exception e){
498-
// System.err.println("Error for " + da_entries.size() + "idx " + entry.getKey() + " " + entry.getValue());
499-
// e.printStackTrace();
500-
// }
501-
// }
502-
}else{
503-
logger.error("@getDataArray Unexpected answer : " + m.getBody());
504-
}
505-
}
506-
for(Map.Entry<CharSequence, DataArray> c: da_entries.entrySet()){
507-
System.out.println(c.getKey());
508-
}
509-
da_entries.entrySet().stream().filter(Objects::nonNull).forEach(a -> System.out.println("> " + a));
510-
List<DataArray> da_list = da_entries.entrySet().stream().filter(Objects::nonNull).sorted(
511-
Comparator.comparingInt(a -> Integer.getInteger(String.valueOf(a.getKey()))))
512-
.map(Map.Entry::getValue).collect(Collectors.toList());
513-
514-
List<Integer> not_found = IntStream.range(0, datasets_paths.size()).filter( i -> !da_entries.containsKey(String.valueOf(i))).boxed().collect(Collectors.toList());
515-
for(Integer i: not_found){
516-
System.out.println("NOT FOUND : " + i);
517-
}
518-
519-
for(DataArray da: da_list){
520-
try {
521-
result.addAll((Collection<? extends Number>) ETPUtils.getAttributeValue(da.getData().getItem(), "value"));
522-
} catch (Exception e) {
523-
logger.error(e);
524-
}
525-
}
526-
return result;*/
527-
528470
}
529471

530472

‎src/main/java/com/geosiris/etp/websocket/ETPClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -342,7 +342,7 @@ public void onWebSocketError(Throwable cause) {
342342
super.onWebSocketError(cause);
343343
logger.debug("@onWebSocketError error ");
344344
if(cause instanceof ConnectException) {
345-
logger.error("ETPClient connection exception");
345+
logger.error("ETPClient connection exception {}", cause);
346346
try {
347347
wsclient.stop();
348348
} catch (Exception e) { logger.error(e.getMessage()); logger.debug(e.getMessage(), e); }

‎src/main/resources/log4j2.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ appender.console.name = STDOUT
66
appender.console.layout.type = PatternLayout
77
appender.console.layout.pattern = [%-5level] %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} - %msg%n
88
appender.console.filter.threshold.type = ThresholdFilter
9-
appender.console.filter.threshold.level = info
9+
appender.console.filter.threshold.level = debug
1010

1111
appender.file.type = File
1212
appender.file.name = LOGFILE

0 commit comments

Comments
 (0)