first full implementation of push of storage configuration and schema

client/src/main/java/com/orientechnologies/orient/client/binary/OBinaryRequestExecutor.java

@@ -117,7 +117,7 @@ public interface OBinaryRequestExecutor {
 
   OBinaryResponse executeSubscribe(OSubscribeRequest request);
 
-  OBinaryResponse executeSubscribePushRequest(OSubscribeDistributedConfigurationRequest request);
+  OBinaryResponse executeSubscribeDistributedConfiguration(OSubscribeDistributedConfigurationRequest request);
 
   OBinaryResponse executeSubscribeLiveQuery(OSubscribeLiveQueryRequest request);
 
@@ -126,4 +126,8 @@ public interface OBinaryRequestExecutor {
   OBinaryResponse executeUnsubscribeLiveQuery(OUnsubscribeLiveQueryRequest request);
 
   OBinaryResponse executeDistributedConnect(ODistributedConnectRequest request);
+
+  OBinaryResponse executeSubscribeStorageConfiguration(OSubscribeStorageConfigurationRequest request);
+
+  OBinaryResponse executeSubscribeSchemaConfiguration(OSubscribeSchemaRequest request);
 }

client/src/main/java/com/orientechnologies/orient/client/remote/ORemotePushHandler.java

@@ -1,9 +1,7 @@
 package com.orientechnologies.orient.client.remote;
 
-import com.orientechnologies.orient.client.remote.message.OBinaryPushRequest;
-import com.orientechnologies.orient.client.remote.message.OBinaryPushResponse;
-import com.orientechnologies.orient.client.remote.message.OLiveQueryPushRequest;
-import com.orientechnologies.orient.client.remote.message.OPushDistributedConfigurationRequest;
+import com.orientechnologies.orient.client.remote.message.*;
+import com.orientechnologies.orient.client.remote.message.push.OStorageConfigurationPayload;
 import com.orientechnologies.orient.enterprise.channel.binary.OChannelBinary;
 
 /**
@@ -17,11 +15,15 @@ public interface ORemotePushHandler {
 
   OBinaryPushResponse executeUpdateDistributedConfig(OPushDistributedConfigurationRequest request);
 
+  OBinaryPushResponse executeUpdateStorageConfig(OPushStorageConfigurationRequest request);
+
   void executeLiveQueryPush(OLiveQueryPushRequest pushRequest);
 
   void onPushReconnect(String host);
 
   void onPushDisconnect(OChannelBinary network, Exception e);
 
   void returnSocket(OChannelBinary network);
+
+  OBinaryPushResponse executeUpdateSchema(OPushSchemaRequest request);
 }

client/src/main/java/com/orientechnologies/orient/client/remote/OStorageConfigurationRemote.java

@@ -1,10 +1,8 @@
 package com.orientechnologies.orient.client.remote;
 
 import com.orientechnologies.orient.client.remote.message.OReloadResponse37;
-import com.orientechnologies.orient.core.config.OContextConfiguration;
-import com.orientechnologies.orient.core.config.OStorageClusterConfiguration;
-import com.orientechnologies.orient.core.config.OStorageConfiguration;
-import com.orientechnologies.orient.core.config.OStorageEntryConfiguration;
+import com.orientechnologies.orient.client.remote.message.push.OStorageConfigurationPayload;
+import com.orientechnologies.orient.core.config.*;
 
 import java.text.SimpleDateFormat;
 import java.util.*;
@@ -36,34 +34,34 @@ public class OStorageConfigurationRemote implements OStorageConfiguration {
   private List<OStorageClusterConfiguration>      clusters;
   private String                                  networkRecordSerializer;
 
-  public OStorageConfigurationRemote(String networkRecordSerializer, OReloadResponse37 response,
+  public OStorageConfigurationRemote(String networkRecordSerializer, OStorageConfigurationPayload payload ,
       OContextConfiguration contextConfiguration) {
     this.networkRecordSerializer = networkRecordSerializer;
     this.contextConfiguration = contextConfiguration;
-    this.dateFormat = response.getDateFormat();
-    this.dateTimeFormat = response.getDateTimeFormat();
-    this.name = response.getName();
-    this.version = response.getVersion();
-    this.directory = response.getDirectory();
+    this.dateFormat = payload.getDateFormat();
+    this.dateTimeFormat = payload.getDateTimeFormat();
+    this.name = payload.getName();
+    this.version = payload.getVersion();
+    this.directory = payload.getDirectory();
     this.properties = new HashMap<>();
-    for (OStorageEntryConfiguration conf : response.getProperties()) {
+    for (OStorageEntryConfiguration conf : payload.getProperties()) {
       this.properties.put(conf.name, conf);
     }
-    this.schemaRecordId = response.getSchemaRecordId().toString();
-    this.indexMgrRecordId = response.getIndexMgrRecordId().toString();
-    this.clusterSelection = response.getClusterSelection();
-    this.conflictStrategy = response.getConflictStrategy();
-    this.validationEnabled = response.isValidationEnabled();
-    this.localeLanguage = response.getLocaleLanguage();
-    this.minimumClusters = response.getMinimumClusters();
-    this.strictSql = response.isStrictSql();
-    this.charset = response.getCharset();
-    this.timeZone = response.getTimeZone();
-    this.localeCountry = response.getLocaleCountry();
-    this.recordSerializer = response.getRecordSerializer();
-    this.recordSerializerVersion = response.getRecordSerializerVersion();
-    this.binaryFormatVersion = response.getBinaryFormatVersion();
-    this.clusters = response.getClusters();
+    this.schemaRecordId = payload.getSchemaRecordId().toString();
+    this.indexMgrRecordId = payload.getIndexMgrRecordId().toString();
+    this.clusterSelection = payload.getClusterSelection();
+    this.conflictStrategy = payload.getConflictStrategy();
+    this.validationEnabled = payload.isValidationEnabled();
+    this.localeLanguage = payload.getLocaleLanguage();
+    this.minimumClusters = payload.getMinimumClusters();
+    this.strictSql = payload.isStrictSql();
+    this.charset = payload.getCharset();
+    this.timeZone = payload.getTimeZone();
+    this.localeCountry = payload.getLocaleCountry();
+    this.recordSerializer = payload.getRecordSerializer();
+    this.recordSerializerVersion = payload.getRecordSerializerVersion();
+    this.binaryFormatVersion = payload.getBinaryFormatVersion();
+    this.clusters = payload.getClusters();
   }
 
   @Override

client/src/main/java/com/orientechnologies/orient/client/remote/OStorageRemote.java

@@ -423,7 +423,7 @@ public OSBTreeCollectionManager getSBtreeCollectionManager() {
   public void reload() {
     OReloadResponse37 res = networkOperation(new OReloadRequest37(), "error loading storage configuration");
     final OStorageConfiguration storageConfiguration = new OStorageConfigurationRemote(
-        ORecordSerializerFactory.instance().getDefaultRecordSerializer().toString(), res, clientConfiguration);
+        ORecordSerializerFactory.instance().getDefaultRecordSerializer().toString(), res.getPayload(), clientConfiguration);
 
     updateStorageConfiguration(storageConfiguration);
   }
@@ -1398,6 +1398,7 @@ private void initPush(OStorageRemoteSession session) {
           pushThread.start();
           subscribeStorageConfiguration(session);
           subscribeDistributedConfiguration(session);
+          subscribeSchema(session);
 
         }
       } finally {
@@ -1411,7 +1412,11 @@ private void subscribeDistributedConfiguration(OStorageRemoteSession nodeSession
   }
 
   private void subscribeStorageConfiguration(OStorageRemoteSession nodeSession) {
-    //TODO
+    pushThread.subscribe(new OSubscribeStorageConfigurationRequest(), nodeSession);
+  }
+
+  private void subscribeSchema(OStorageRemoteSession nodeSession) {
+    pushThread.subscribe(new OSubscribeSchemaRequest(), nodeSession);
   }
 
   protected void openRemoteDatabase(String currentURL) {
@@ -1883,9 +1888,11 @@ public OBinaryPushRequest createPush(byte type) {
       return new OPushDistributedConfigurationRequest();
     case OChannelBinaryProtocol.REQUEST_PUSH_LIVE_QUERY:
       return new OLiveQueryPushRequest();
-//    case OChannelBinaryProtocol.REQUEST_PUSH_STORAGE_CONFIG:
-//
-//      return  new
+    case OChannelBinaryProtocol.REQUEST_PUSH_STORAGE_CONFIG:
+      return new OPushStorageConfigurationRequest();
+    case OChannelBinaryProtocol.REQUEST_PUSH_SCHEMA:
+      return new OPushSchemaRequest();
+
     }
     return null;
   }
@@ -1896,6 +1903,21 @@ public OBinaryPushResponse executeUpdateDistributedConfig(OPushDistributedConfig
     return null;
   }
 
+  @Override
+  public OBinaryPushResponse executeUpdateStorageConfig(OPushStorageConfigurationRequest payload) {
+    final OStorageConfiguration storageConfiguration = new OStorageConfigurationRemote(
+        ORecordSerializerFactory.instance().getDefaultRecordSerializer().toString(), payload.getPayload(), clientConfiguration);
+
+    updateStorageConfiguration(storageConfiguration);
+    return null;
+  }
+
+  @Override
+  public OBinaryPushResponse executeUpdateSchema(OPushSchemaRequest request) {
+    ODatabaseDocumentRemote.updateSchema(this, request.getSchema());
+    return null;
+  }
+
   public OLiveQueryMonitor liveQuery(ODatabaseDocumentRemote database, String query, OLiveQueryClientListener listener,
       Object[] params) {
 

client/src/main/java/com/orientechnologies/orient/client/remote/message/OPushSchemaRequest.java

@@ -0,0 +1,57 @@
+package com.orientechnologies.orient.client.remote.message;
+
+import com.orientechnologies.orient.client.remote.ORemotePushHandler;
+import com.orientechnologies.orient.client.remote.message.push.OStorageConfigurationPayload;
+import com.orientechnologies.orient.core.config.OStorageConfiguration;
+import com.orientechnologies.orient.core.record.impl.ODocument;
+import com.orientechnologies.orient.core.serialization.serializer.record.binary.ORecordSerializerNetworkV37;
+import com.orientechnologies.orient.enterprise.channel.binary.OChannelDataInput;
+import com.orientechnologies.orient.enterprise.channel.binary.OChannelDataOutput;
+
+import java.io.IOException;
+
+import static com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryProtocol.REQUEST_PUSH_SCHEMA;
+import static com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryProtocol.REQUEST_PUSH_STORAGE_CONFIG;
+
+public class OPushSchemaRequest implements OBinaryPushRequest<OBinaryPushResponse> {
+
+  private ODocument schema;
+
+  public OPushSchemaRequest() {
+
+  }
+
+  public OPushSchemaRequest(ODocument schema) {
+    this.schema = schema;
+  }
+
+  @Override
+  public void write(OChannelDataOutput channel) throws IOException {
+    channel.writeBytes(ORecordSerializerNetworkV37.INSTANCE.toStream(schema, false));
+  }
+
+  @Override
+  public void read(OChannelDataInput network) throws IOException {
+    byte[] bytes = network.readBytes();
+    this.schema = (ODocument) ORecordSerializerNetworkV37.INSTANCE.fromStream(bytes, null, null);
+  }
+
+  @Override
+  public OBinaryPushResponse execute(ORemotePushHandler pushHandler) {
+    return pushHandler.executeUpdateSchema(this);
+  }
+
+  @Override
+  public OBinaryPushResponse createResponse() {
+    return null;
+  }
+
+  @Override
+  public byte getPushCommand() {
+    return REQUEST_PUSH_SCHEMA;
+  }
+
+  public ODocument getSchema() {
+    return schema;
+  }
+}

client/src/main/java/com/orientechnologies/orient/client/remote/message/OPushStorageConfigurationRequest.java

@@ -0,0 +1,56 @@
+package com.orientechnologies.orient.client.remote.message;
+
+import com.orientechnologies.orient.client.remote.ORemotePushHandler;
+import com.orientechnologies.orient.client.remote.message.OBinaryPushRequest;
+import com.orientechnologies.orient.client.remote.message.OBinaryPushResponse;
+import com.orientechnologies.orient.client.remote.message.OPushDistributedConfigurationRequest;
+import com.orientechnologies.orient.client.remote.message.push.OStorageConfigurationPayload;
+import com.orientechnologies.orient.core.config.OStorageConfiguration;
+import com.orientechnologies.orient.enterprise.channel.binary.OChannelDataInput;
+import com.orientechnologies.orient.enterprise.channel.binary.OChannelDataOutput;
+
+import java.io.IOException;
+
+import static com.orientechnologies.orient.enterprise.channel.binary.OChannelBinaryProtocol.REQUEST_PUSH_STORAGE_CONFIG;
+
+public class OPushStorageConfigurationRequest implements OBinaryPushRequest<OBinaryPushResponse> {
+
+  private OStorageConfigurationPayload payload;
+
+  public OPushStorageConfigurationRequest() {
+    payload = new OStorageConfigurationPayload();
+  }
+
+  public OPushStorageConfigurationRequest(OStorageConfiguration configuration) {
+    payload = new OStorageConfigurationPayload(configuration);
+  }
+
+  @Override
+  public void write(OChannelDataOutput channel) throws IOException {
+    payload.write(channel);
+  }
+
+  @Override
+  public void read(OChannelDataInput network) throws IOException {
+    payload.read(network);
+  }
+
+  @Override
+  public OBinaryPushResponse execute(ORemotePushHandler pushHandler) {
+    return pushHandler.executeUpdateStorageConfig(this);
+  }
+
+  @Override
+  public OBinaryPushResponse createResponse() {
+    return null;
+  }
+
+  @Override
+  public byte getPushCommand() {
+    return REQUEST_PUSH_STORAGE_CONFIG;
+  }
+
+  public OStorageConfigurationPayload getPayload() {
+    return payload;
+  }
+}