[WIP]merge develop into master

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/ClientOffPublishers.java

@@ -18,14 +18,15 @@
 
 import com.alipay.sofa.registry.common.model.store.Publisher;
 import com.google.common.collect.Lists;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
 public class ClientOffPublishers {
   private final ConnectId connectId;
   private final List<Publisher> publishers;
 
-  public ClientOffPublishers(ConnectId connectId, List<Publisher> publishers) {
+  public ClientOffPublishers(ConnectId connectId, Collection<Publisher> publishers) {
     this.connectId = connectId;
     this.publishers = Collections.unmodifiableList(Lists.newArrayList(publishers));
   }

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/dataserver/ClientOffPublisher.java

@@ -29,13 +29,15 @@
  * @author qian.lqlq
  * @version $Id: ClientOffPublisher.java, v 0.1 2017-12-01 15:48 qian.lqlq Exp $
  */
-public final class ClientOffPublisher implements Serializable {
+public final class ClientOffPublisher implements Serializable, DataServerReq {
   private static final long serialVersionUID = -3547806571058756207L;
 
+  private final String dataInfoId; // use to cal slot id
   private final ConnectId connectId;
   private final Map<String, Map<String, RegisterVersion>> publisherMap = Maps.newHashMap();
 
-  public ClientOffPublisher(ConnectId connectId) {
+  public ClientOffPublisher(String dataInfoId, ConnectId connectId) {
+    this.dataInfoId = dataInfoId;
     this.connectId = connectId;
   }
 
@@ -61,4 +63,9 @@ public boolean isEmpty() {
   public String toString() {
     return "ClientOff{" + "connId=" + connectId + ", pubs=" + publisherMap + '}';
   }
+
+  @Override
+  public String getDataInfoId() {
+    return dataInfoId;
+  }
 }

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/wrapper/Wrapper.java → server/common/model/src/main/java/com/alipay/sofa/registry/common/model/dataserver/DataServerReq.java

@@ -14,17 +14,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.alipay.sofa.registry.common.model.wrapper;
+package com.alipay.sofa.registry.common.model.dataserver;
 
-import java.util.function.Supplier;
+/** Request to write to DataNode. */
+public interface DataServerReq {
 
-/**
- * @author shangyu.wh
- * @version 1.0: Wrapper.java, v 0.1 2019-06-18 19:33 shangyu.wh Exp $
- */
-public interface Wrapper<T, R> {
-
-  R call();
-
-  Supplier<T> getParameterSupplier();
+  /**
+   * Info id of data.
+   *
+   * @return data info id
+   */
+  String getDataInfoId();
 }

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/BaseInfo.java

@@ -18,6 +18,7 @@
 
 import com.alipay.sofa.registry.common.model.ConnectId;
 import com.alipay.sofa.registry.common.model.RegisterVersion;
+import com.alipay.sofa.registry.common.model.dataserver.DataServerReq;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.google.common.collect.Maps;
 import java.io.Serializable;
@@ -30,7 +31,7 @@
  * @author shangyu.wh
  * @version $Id: BaseInfo.java, v 0.1 2017-11-30 16:31 shangyu.wh Exp $
  */
-public abstract class BaseInfo implements Serializable, StoreData<String> {
+public abstract class BaseInfo implements Serializable, StoreData<String>, DataServerReq {
 
   private static final long serialVersionUID = -6263388188316303789L;
 

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/StoreData.java

@@ -16,6 +16,9 @@
  */
 package com.alipay.sofa.registry.common.model.store;
 
+import com.alipay.sofa.registry.common.model.ConnectId;
+import com.alipay.sofa.registry.common.model.RegisterVersion;
+
 /**
  * @author shangyu.wh
  * @version $Id: StoreData.java, v 0.1 2017-11-30 19:48 shangyu.wh Exp $
@@ -43,4 +46,12 @@ enum DataType {
    * @return
    */
   ID getId();
+
+  String getDataInfoId();
+
+  RegisterVersion registerVersion();
+
+  long getRegisterTimestamp();
+
+  ConnectId connectId();
 }

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/Subscriber.java

@@ -69,7 +69,7 @@ public ElementType getElementType() {
   }
 
   private PushContext getPushContext(String dataCenter) {
-    PushContext ctx = null;
+    PushContext ctx;
     if (lastPushContexts == null) {
       ctx = new PushContext();
       this.lastPushContexts = Collections.singletonMap(dataCenter, ctx);
@@ -223,7 +223,6 @@ public synchronized long markPushEmpty(String dataCenter, long emptyVersion) {
     return emptyVersion;
   }
 
-  /** @return */
   public synchronized CircuitBreakerStatistic getStatistic(String dataCenter) {
     final PushContext ctx = getPushContext(dataCenter);
     return new CircuitBreakerStatistic(
@@ -234,28 +233,6 @@ public synchronized CircuitBreakerStatistic getStatistic(String dataCenter) {
         ctx.lastPushedFailTimeStamp);
   }
 
-  /**
-   * change subscriber word cache
-   *
-   * @param subscriber
-   * @return
-   */
-  public static Subscriber internSubscriber(Subscriber subscriber) {
-    subscriber.setDataInfoId(subscriber.getDataInfoId());
-    subscriber.setInstanceId(subscriber.getInstanceId());
-    subscriber.setGroup(subscriber.getGroup());
-    subscriber.setDataId(subscriber.getDataId());
-    subscriber.setCell(subscriber.getCell());
-    subscriber.setProcessId(subscriber.getProcessId());
-    subscriber.setAppName(subscriber.getAppName());
-
-    subscriber.setSourceAddress(URL.internURL(subscriber.getSourceAddress()));
-    subscriber.setTargetAddress(URL.internURL(subscriber.getTargetAddress()));
-    subscriber.setAttributes(subscriber.getAttributes());
-
-    return subscriber;
-  }
-
   protected Map<String, String> internAttributes(Map<String, String> attributes) {
     Map<String, String> intern = super.internAttributes(attributes);
     return com.alipay.sofa.registry.collections.Maps.trimMap(intern);

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/Watcher.java

@@ -22,6 +22,7 @@
  */
 public class Watcher extends BaseInfo {
 
+  private static final long serialVersionUID = -2122505760402915804L;
   private volatile long pushedVersion;
 
   @Override
@@ -53,27 +54,4 @@ public String shortDesc() {
     sb.append("sourceAddress=").append(getSourceAddress().buildAddressString());
     return sb.toString();
   }
-
-  /**
-   * change watcher word cache
-   *
-   * @param watcher
-   * @return
-   */
-  public static Watcher internWatcher(Watcher watcher) {
-    watcher.setRegisterId(watcher.getRegisterId());
-    watcher.setDataInfoId(watcher.getDataInfoId());
-    watcher.setInstanceId(watcher.getInstanceId());
-    watcher.setGroup(watcher.getGroup());
-    watcher.setDataId(watcher.getDataId());
-    watcher.setClientId(watcher.getClientId());
-    watcher.setCell(watcher.getCell());
-    watcher.setProcessId(watcher.getProcessId());
-    watcher.setAppName(watcher.getAppName());
-    watcher.setSourceAddress(URL.internURL(watcher.getSourceAddress()));
-    watcher.setTargetAddress(URL.internURL(watcher.getTargetAddress()));
-    watcher.setAttributes(watcher.getAttributes());
-
-    return watcher;
-  }
 }

server/common/util/src/main/java/com/alipay/sofa/registry/exception/InterceptorExecutionException.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alipay.sofa.registry.exception;
+
+/** Interceptor exception */
+public class InterceptorExecutionException extends RuntimeException {
+
+  private static final long serialVersionUID = -6793552645172892864L;
+
+  public InterceptorExecutionException(String message) {
+    super(message);
+  }
+
+  public InterceptorExecutionException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

server/server/data/src/test/java/com/alipay/sofa/registry/server/data/remoting/sessionserver/handler/BatchPutDataHandlerTest.java

@@ -127,7 +127,7 @@ private static BatchRequest request(ConnectId connectId, boolean illegalArg) {
     Publisher pub1 = TestBaseUtils.createTestPublisher("testDataId");
     Publisher pub2 = TestBaseUtils.createTestPublisher("testDataId");
 
-    ClientOffPublisher off = new ClientOffPublisher(connectId);
+    ClientOffPublisher off = new ClientOffPublisher(pub2.getDataInfoId(), connectId);
     off.addPublisher(pub2);
     Assert.assertFalse(off.isEmpty());
     Assert.assertEquals(off.getConnectId(), connectId);

server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/cleaner/AppRevisionCleaner.java

@@ -162,30 +162,30 @@ void markDeleted() {
 
   private void appRevisionSwitchRefresh() {
     DBResponse<PersistenceData> ret =
-            provideDataService.queryProvideData(ValueConstants.APP_REVISION_WRITE_SWITCH_DATA_ID);
+        provideDataService.queryProvideData(ValueConstants.APP_REVISION_WRITE_SWITCH_DATA_ID);
     AppRevisionDomainConvertor.EnableConfig enableConfig = null;
     if (ret.getOperationStatus() == OperationStatus.SUCCESS) {
       PersistenceData data = ret.getEntity();
       String switchString = data.getData();
       if (StringUtils.isNotBlank(switchString)) {
         try {
           enableConfig =
-                  JsonUtils.read(switchString, AppRevisionDomainConvertor.EnableConfig.class);
+              JsonUtils.read(switchString, AppRevisionDomainConvertor.EnableConfig.class);
         } catch (Throwable e) {
           LOG.error("Decode appRevision write switch failed", e);
         }
       }
     }
     if (enableConfig != null) {
       LOG.info(
-              "appRevisionSwitch prev={}/{}",
-              AppRevisionDomainConvertor.getEnableConfig().isServiceParams(),
-              AppRevisionDomainConvertor.getEnableConfig().isServiceParamsLarge());
+          "appRevisionSwitch prev={}/{}",
+          AppRevisionDomainConvertor.getEnableConfig().isServiceParams(),
+          AppRevisionDomainConvertor.getEnableConfig().isServiceParamsLarge());
       AppRevisionDomainConvertor.setEnableConfig(enableConfig);
       LOG.info(
-              "appRevisionSwitch update={}/{}",
-              enableConfig.isServiceParams(),
-              enableConfig.isServiceParamsLarge());
+          "appRevisionSwitch update={}/{}",
+          enableConfig.isServiceParams(),
+          enableConfig.isServiceParamsLarge());
     }
   }
 

server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/cleaner/AppRevisionCleanerTest.java

@@ -69,11 +69,12 @@ public void beforeTest() throws Exception {
     // doReturn(new DateNowDomain(new Date())).when(appRevisionCleaner.appRevisionMapper).getNow();
     doReturn(
             new DBResponse<>(
-                    PersistenceDataBuilder.createPersistenceData(
-                            ValueConstants.APP_REVISION_WRITE_SWITCH_DATA_ID, "{\"serviceParams\":false,\"serviceParamsLarge\":true}"),
-                    OperationStatus.SUCCESS))
-            .when(appRevisionCleaner.provideDataService)
-            .queryProvideData(anyString());
+                PersistenceDataBuilder.createPersistenceData(
+                    ValueConstants.APP_REVISION_WRITE_SWITCH_DATA_ID,
+                    "{\"serviceParams\":false,\"serviceParamsLarge\":true}"),
+                OperationStatus.SUCCESS))
+        .when(appRevisionCleaner.provideDataService)
+        .queryProvideData(anyString());
   }
 
   @After

server/server/session/src/main/java/com/alipay/sofa/registry/server/session/acceptor/ClientOffWriteDataRequest.java

@@ -19,18 +19,18 @@
 import com.alipay.sofa.registry.common.model.ClientOffPublishers;
 import com.alipay.sofa.registry.common.model.ConnectId;
 import com.alipay.sofa.registry.common.model.store.Publisher;
-import java.util.List;
+import java.util.Collection;
 
 /**
  * @author yuzhi.lyz
  * @version v 0.1 2020-12-12 15:20 yuzhi.lyz Exp $
  */
 public final class ClientOffWriteDataRequest implements WriteDataRequest<ClientOffPublishers> {
-  private final ConnectId connectId;
 
+  private final ConnectId connectId;
   private final ClientOffPublishers requestBody;
 
-  public ClientOffWriteDataRequest(ConnectId connectId, List<Publisher> publishers) {
+  public ClientOffWriteDataRequest(ConnectId connectId, Collection<Publisher> publishers) {
     this.connectId = connectId;
     this.requestBody = new ClientOffPublishers(connectId, publishers);
   }

server/server/session/src/main/java/com/alipay/sofa/registry/server/session/acceptor/PublisherWriteDataRequest.java → server/server/session/src/main/java/com/alipay/sofa/registry/server/session/acceptor/PublisherRegisterWriteDataRequest.java

@@ -19,17 +19,13 @@
 import com.alipay.sofa.registry.common.model.ConnectId;
 import com.alipay.sofa.registry.common.model.store.Publisher;
 
-/**
- * @author yuzhi.lyz
- * @version v 0.1 2020-12-12 15:14 yuzhi.lyz Exp $
- */
-public final class PublisherWriteDataRequest implements WriteDataRequest<Publisher> {
+/** WriteDataRequest of publisher register. */
+public final class PublisherRegisterWriteDataRequest implements WriteDataRequest<Publisher> {
+
   private final Publisher publisher;
-  private final WriteDataRequestType type;
 
-  public PublisherWriteDataRequest(Publisher publisher, WriteDataRequestType type) {
+  public PublisherRegisterWriteDataRequest(Publisher publisher) {
     this.publisher = publisher;
-    this.type = type;
   }
 
   @Override
@@ -39,7 +35,7 @@ public Publisher getRequestBody() {
 
   @Override
   public WriteDataRequestType getRequestType() {
-    return type;
+    return WriteDataRequestType.PUBLISHER;
   }
 
   @Override