add function sf_proto_deal_ack

src/idempotency/client/receipt_handler.c

@@ -278,11 +278,6 @@ static int deal_setup_channel_response(struct fast_task_info *task)
     }
     channel->buffer_size = FC_MIN(buffer_size, task->size);
 
-    logInfo("file: "__FILE__", line: %d, "
-            "peer buffer size: %d, mine buffer size: %d, "
-            "min buffer size: %d", __LINE__, buffer_size,
-            task->size, channel->buffer_size);
-
     PTHREAD_MUTEX_LOCK(&channel->lc_pair.lock);
     pthread_cond_broadcast(&channel->lc_pair.cond);
     PTHREAD_MUTEX_UNLOCK(&channel->lc_pair.lock);

src/idempotency/client/rpc_wrapper.h

@@ -64,11 +64,11 @@
     \
             SF_NET_RETRY_CHECK_AND_SLEEP(net_retry_ctx, client_ctx-> \
                     net_retry_cfg.network.times, ++i, result); \
-    \
+    /* \
             logInfo("file: "__FILE__", line: %d, func: %s, "  \
                     "net retry result: %d, retry count: %d",  \
                     __LINE__, __FUNCTION__, result, i);       \
-    \
+    */ \
             SF_CLIENT_RELEASE_CONNECTION(client_ctx, conn, result); \
             if ((conn=GET_MASTER_CONNECTION(client_ctx,  \
                             get_conn_arg1, &result)) == NULL)  \
@@ -128,11 +128,11 @@
         }  \
         SF_NET_RETRY_CHECK_AND_SLEEP(net_retry_ctx, client_ctx-> \
                 net_retry_cfg.network.times, ++i, result); \
-    \
+    /* \
         logInfo("file: "__FILE__", line: %d, func: %s, "  \
                 "net retry result: %d, retry count: %d",  \
                 __LINE__, __FUNCTION__, result, i);       \
-    \
+    */ \
         SF_CLIENT_RELEASE_CONNECTION(client_ctx, conn, result); \
         if ((conn=GET_READABLE_CONNECTION(client_ctx,  \
                         get_conn_arg1, &result)) == NULL)  \

src/idempotency/server/server_channel.c

@@ -290,8 +290,6 @@ void idempotency_channel_free(IdempotencyChannel *channel)
 
 int idempotency_request_alloc_init(void *element, void *args)
 {
-    static int i = 0;
-
     IdempotencyRequest *request;
     request = (IdempotencyRequest *)element;
     request->allocator = (struct fast_mblock_man *)args;
@@ -299,11 +297,5 @@ int idempotency_request_alloc_init(void *element, void *args)
         sizeof(IdempotencyRequest);
     request->output.response = request + 1;
 
-    if (++i % 100 == 0) {
-        logInfo("i: %d, element_size: %d, rsize: %d", i,
-                request->allocator->info.element_size,
-                request->output.rsize);
-    }
-
     return 0;
 }

src/idempotency/server/server_handler.c

@@ -153,8 +153,7 @@ int sf_server_deal_report_req_receipt(struct fast_task_info *task,
         }
     }
 
-    logInfo("receipt count: %d, success: %d", count, success);
-
+    //logInfo("receipt count: %d, success: %d", count, success);
     response->header.cmd = SF_SERVICE_PROTO_REPORT_REQ_RECEIPT_RESP;
     return 0;
 }

src/sf_proto.c

@@ -240,3 +240,40 @@ const char *sf_get_cmd_caption(const int cmd)
             return "UNKOWN";
     }
 }
+
+int sf_proto_deal_ack(struct fast_task_info *task,
+        SFRequestInfo *request, SFResponseInfo *response)
+{
+    if (request->header.status != 0) {
+        if (request->header.body_len > 0) {
+            int remain_size;
+            int len;
+
+            response->error.length = sprintf(response->error.message,
+                    "message from peer %s:%u => ",
+                    task->client_ip, task->port);
+            remain_size = sizeof(response->error.message) -
+                response->error.length;
+            if (request->header.body_len >= remain_size) {
+                len = remain_size - 1;
+            } else {
+                len = request->header.body_len;
+            }
+
+            memcpy(response->error.message + response->error.length,
+                    request->body, len);
+            response->error.length += len;
+            *(response->error.message + response->error.length) = '\0';
+        }
+
+        return request->header.status;
+    }
+
+    if (request->header.body_len > 0) {
+        response->error.length = sprintf(response->error.message,
+                "ACK body length: %d != 0", request->header.body_len);
+        return -EINVAL;
+    }
+
+    return 0;
+}

src/sf_proto.h

@@ -254,6 +254,16 @@ static inline int sf_active_test(ConnectionInfo *conn,
             SF_PROTO_ACTIVE_TEST_RESP);
 }
 
+static inline int sf_proto_deal_actvie_test(struct fast_task_info *task,
+        SFRequestInfo *request, SFResponseInfo *response)
+{
+    return sf_server_expect_body_length(response,
+            request->header.body_len, 0);
+}
+
+int sf_proto_deal_ack(struct fast_task_info *task,
+        SFRequestInfo *request, SFResponseInfo *response);
+
 #define SF_CLIENT_RELEASE_CONNECTION(client_ctx, conn, result) \
     do {  \
         if (SF_FORCE_CLOSE_CONNECTION_ERROR(result)) {  \