Merge pull request 'feat-protobuf' (#1) from feat-protobuf into master

Reviewed-on: #1
2023-12-30 14:24:35 +05:30 · 2023-12-30 14:24:35 +05:30 · b4469e03d0
parent aed464fe38 01d4c2fce3
commit b4469e03d0
39 changed files with 4189 additions and 1237 deletions
--- a/.woodpecker.yml
+++ b/.woodpecker.yml
@ -2,6 +2,8 @@ steps:
  backend:
    image: rust
    commands:
+      - apt update
+      - apt-get install -y --no-install-recommends protobuf-compiler
      - cargo build
      #      - make migrate
      #      - make
--- a/Cargo.lock
+++ b/Cargo.lock
--- a/Cargo.toml
+++ b/Cargo.toml
@ -2,6 +2,7 @@
 name = "dcache"
 version = "0.1.0"
 edition = "2021"
+build = "build.rs"

 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

@ -15,30 +16,31 @@ tracing = { version = "0.1.37", features = ["log"] }
 serde_json = "1.0.96"
 serde = { version = "1.0.163", features = ["derive"] }
 byteorder = "1.4.3"
-actix-web = "4"
-actix-web-httpauth = "0.8.0"
 futures-util = { version = "0.3.17", default-features = false, features = ["std"] }
 lazy_static = "1.4.0"
 pretty_env_logger = "0.4.0"
 uuid = { version = "1", features = ["v4"] }
-actix-web-codegen-const-routes = { version = "0.1.0", tag = "0.1.0", git = "https://github.com/realaravinth/actix-web-codegen-const-routes" }
 derive_builder = "0.11.2"
 config = { version = "0.11", features = ["toml"] }
 derive_more = "0.99.17"
 url = { version = "2.2.2", features = ["serde"]}
 async-trait = "0.1.36"
 clap = { version = "4.1.11", features = ["derive", "env"] }
-reqwest = { version = "0.11.9", features = ["json"] }
-tokio = { version = "1.0", default-features = false, features = ["sync"] }
+tokio = { version = "1.0", default-features = false, features = ["sync", "macros", "rt-multi-thread"] }
 tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
 actix = "0.13.0"
+tonic = { version = "0.10.2", features = ["transport", "channel"] }
+prost = "0.12.3"
+tokio-stream = "0.1.14"
+async-stream = "0.3.5"
+actix-rt = "2.9.0"


 [build-dependencies]
 serde_json = "1"
+tonic-build = "0.10.2"

 [dev-dependencies]
-actix-rt = "2.7.0"
 base64 = "0.13.0"
 anyhow = "1.0.63"
 maplit = "1.0.2"
--- a/1
+++ b/1
@ -4,6 +4,7 @@

 FROM rust:latest as rust
 WORKDIR /src
+RUN apt update && apt-get install -y --no-install-recommends protobuf-compiler
 COPY . .
 RUN cargo build --release

--- a/6
+++ b/6
@ -0,0 +1,6 @@
+python.gen:
+	. venv/bin/activate && cd dcache_py
+	python -m grpc_tools.protoc \
+		-I=./proto/dcache/ --python_out=dcache_py/ \
+		--pyi_out=dcache_py/ \
+		--grpc_python_out=dcache_py/ ./proto/dcache/dcache.proto
--- a/bench/adaptor.py
+++ b/bench/adaptor.py
@ -0,0 +1,64 @@
+import time
+from typing import Any, Callable
+import grpc
+import grpc.experimental.gevent as grpc_gevent
+from grpc_interceptor import ClientInterceptor
+from locust import User
+from locust.exception import LocustError
+
+# patch grpc so that it uses gevent instead of asyncio
+grpc_gevent.init_gevent()
+
+
+class LocustInterceptor(ClientInterceptor):
+    def __init__(self, environment, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self.env = environment
+
+    def intercept(
+        self,
+        method: Callable,
+        request_or_iterator: Any,
+        call_details: grpc.ClientCallDetails,
+    ):
+        response = None
+        exception = None
+        start_perf_counter = time.perf_counter()
+        response_length = 0
+        try:
+            response = method(request_or_iterator, call_details)
+            response_length = response.result().ByteSize()
+        except grpc.RpcError as e:
+            exception = e
+
+        self.env.events.request.fire(
+            request_type="grpc",
+            name=call_details.method,
+            response_time=(time.perf_counter() - start_perf_counter) * 1000,
+            response_length=response_length,
+            response=response,
+            context=None,
+            exception=exception,
+        )
+        return response
+
+
+class GrpcUser(User):
+    abstract = True
+    stub_class = None
+
+    def __init__(self, environment):
+        super().__init__(environment)
+        for attr_value, attr_name in (
+            (self.host, "host"),
+            (self.stub_class, "stub_class"),
+        ):
+            if attr_value is None:
+                raise LocustError(f"You must specify the {attr_name}.")
+
+        self._channel = grpc.insecure_channel(self.host)
+        interceptor = LocustInterceptor(environment=environment)
+        self._channel = grpc.intercept_channel(self._channel, interceptor)
+
+        self.stub = self.stub_class(self._channel)
--- a/bench/locustfile.py
+++ b/bench/locustfile.py
@ -1,95 +1,69 @@
+import json
+import time
+import grpc
+import gevent
 from pprint import pprint
-from locust import FastHttpUser, between, task
+from locust import FastHttpUser, between, task, events

-password = "fooobarasdfasdf"
-username = "realaravinth"
+from dcache_py import dcache_pb2 as dcache
+from dcache_py.dcache_pb2 import RaftRequest
+from dcache_py.dcache_pb2_grpc import DcacheServiceStub
+
+import adaptor
+
+host = "localhost:9001"
+captcha_id = "locust"


-class Unprotected(FastHttpUser):
-    #    wait_time = between(5, 15)
-    peers = [
-            "http://localhost:9001",
-            "http://localhost:9002",
-            "http://localhost:9003",
-            ]
-    leader = "http://localhost:9001"
-    host = leader
-    captcha_id="locust"
-
-    pipeline_vote = []
-    for _ in range(0,1000):
-        pipeline_vote.append({"AddVisitor": captcha_id})
-
-#    def on_start(self):
-#        resp  = self.client.get(f"{self.leader}/metrics")
-#        data = resp.json()
-#        leader = data["Ok"]["membership_config"]["log_id"]["leader_id"]["node_id"]
-#        self.leader = self.peers[leader - 1]
-#        self.host = self.leader
-#        print(f"Leader: {self.host}")
-#        self.add_captcha(captcha_id="locust")
+def add_captcha(stub: DcacheServiceStub, captcha_id: str):
+    msg = dcache.AddCaptchaRequest(
+        id=captcha_id,
+        mcaptcha=dcache.MCaptcha(
+            duration=30,
+            defense=dcache.Defense(
+                levels=[
+                    dcache.Level(visitor_threshold=50, difficulty_factor=500),
+                    dcache.Level(visitor_threshold=5000, difficulty_factor=50000),
+                ]
+            ),
+        ),
+    )
+    resp = stub.AddCaptcha(msg)
+    pprint(f"Captcha added {captcha_id}: {resp}")


-
-    def write(self, data):
-        resp = self.client.post(f"{self.host}/write", json=data)
-        print(f"RPC Status: {resp.status_code}")
-        resp = resp.json()
-        if "Err" in resp:
-            leader = resp["Err"]["APIError"]["ForwardToLeader"]["leader_node"]["addr"]
-            print(f"Forwarding write to leader {leader}")
-            return write(leader, data)
-        return resp["Ok"]["data"]
-
-    def pipeline_write(self, data):
-        resp = self.client.post(f"{self.host}/pipeline/write", json=data)
-#        print(f"RPC Status: {resp.status_code}")
-        resp = resp.json()
-        if "Err" in resp:
-            leader = resp["Err"]["APIError"]["ForwardToLeader"]["leader_node"]["addr"]
-            print(f"Forwarding write to leader {leader}")
-            return write(leader, data)
-        return resp
+with grpc.insecure_channel(host) as channel:
+    stub = DcacheServiceStub(channel)
+    add_captcha(stub=stub, captcha_id=captcha_id)


+pipeline_msgs = []
+for _ in range(0,10):
+    pipeline_msgs.append(dcache.DcacheRequest(addVisitor=dcache.CaptchaID(id=captcha_id)))
+pipeline_msgs = dcache.DcacheBatchRequest(requests=pipeline_msgs)
+
+#def pipeline_generate_messages():
+#    for msg in pipeline_msgs:
+#        yield msg
+
+
+class HelloGrpcUser(adaptor.GrpcUser):
+    stub_class = DcacheServiceStub
+    host = host
+    captcha_id = captcha_id
+    msg = dcache.CaptchaID(id=captcha_id)


    def add_vote(self, captcha_id: str):
-        resp = self.write(data={"AddVisitor": captcha_id})
-        pprint(resp)
+        resp = self.stub.AddVisitor(self.msg)

-    def add_vote_pipeline(self, captcha_id: str):
-        resp = self.pipeline_write(data=self.pipeline_vote)
-#        pprint(resp)
-
-    def add_captcha(self, captcha_id: str):
-        params = {
-            "AddCaptcha": {
-                "id": captcha_id,
-                "mcaptcha": {
-                    "visitor_threshold": 0,
-                    "defense": {
-                        "levels": [
-                            {"visitor_threshold": 50, "difficulty_factor": 500},
-                            {"visitor_threshold": 5000, "difficulty_factor": 50000},
-                        ],
-                        "current_visitor_threshold": 0,
-                    },
-                    "duration": 30,
-                },
-            }
-        }
-        resp = self.write(data=params)
-        pprint(f"Captcha added {captcha_id}: {resp}")
+    def add_vote_pipeline(self):
+        res = self.stub.PipelineDcacheOps(pipeline_msgs)

+#    @task
+#    def addVote(self):
+#        self.add_vote(self.captcha_id)

    @task
-    def unprotected(self):
-        self.add_vote_pipeline(captcha_id=self.captcha_id)
-        ##self.add_vote(captcha_id="locust")
-#        data = {
-#            "username": username,
-#            "password": username,
-#            "confirm_password": username,
-#        }
-#        self.client.post("/unprotected", data=data)
+    def addVotePipeline(self):
+        self.add_vote_pipeline()
--- a/bench/results/v2/grpc-no-conn-pool-single/bench.sh
+++ b/bench/results/v2/grpc-no-conn-pool-single/bench.sh
@ -0,0 +1,7 @@
+#!/bin/bash
+
+ ghz --insecure --proto ./proto/dcache/dcache.proto --call dcache.DcacheService.Write \
+	 -c 300 -n 30000 --rps 4000 -O html -o out.html \
+	 -d '{"data":"{\"AddVisitor\": \"test_1\"}"}' \
+	 localhost:9001
+#	 -d '{"data":"{\"AddCaptcha\":{\"id\":\"test_1\",\"mcaptcha\":{\"defense\":{\"current_visitor_threshold\":0,\"levels\":[{\"difficulty_factor\":500,\"visitor_threshold\":50},{\"difficulty_factor\":50000,\"visitor_threshold\":5000}]},\"duration\":30,\"visitor_threshold\":0}}}"}' \
--- a/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/README.md
+++ b/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/README.md
@ -0,0 +1 @@
+10 messages per batch request 
--- a/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/exceptions_1703846156.5374856.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/exceptions_1703846156.5374856.csv
@ -0,0 +1 @@
+Count,Message,Traceback,Nodes
--- a/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/failures_1703846155.0632446.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/failures_1703846155.0632446.csv
@ -0,0 +1 @@
+Method,Name,Error,Occurrences
--- a/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/report_1703846160.6884449.html
+++ b/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/report_1703846160.6884449.html
--- a/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/requests_1703846136.072447.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/grpc-no-conn-pool-batch/requests_1703846136.072447.csv
@ -0,0 +1,3 @@
+Type,Name,Request Count,Failure Count,Median Response Time,Average Response Time,Min Response Time,Max Response Time,Average Content Size,Requests/s,Failures/s,50%,66%,75%,80%,90%,95%,98%,99%,99.9%,99.99%,100%
+grpc,/dcache.DcacheService/PipelineDcacheOps,41383,0,140.0,99.16818701259079,5.581609002547339,182.89305199868977,40.0,650.896214047811,0.0,140,150,150,150,160,160,160,170,180,180,180
+,Aggregated,41383,0,140.0,99.16818701259079,5.581609002547339,182.89305199868977,40.0,650.896214047811,0.0,140,150,150,150,160,160,160,170,180,180,180
--- a/bench/results/v2/grpc-no-conn-pool-single/json/exceptions_1703750739.8121767.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/json/exceptions_1703750739.8121767.csv
@ -0,0 +1 @@
+Count,Message,Traceback,Nodes
--- a/bench/results/v2/grpc-no-conn-pool-single/json/failures_1703750736.2690804.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/json/failures_1703750736.2690804.csv
@ -0,0 +1 @@
+Method,Name,Error,Occurrences
--- a/bench/results/v2/grpc-no-conn-pool-single/json/report_1703750744.1268215.html
+++ b/bench/results/v2/grpc-no-conn-pool-single/json/report_1703750744.1268215.html
--- a/bench/results/v2/grpc-no-conn-pool-single/json/requests_1703750722.6138687.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/json/requests_1703750722.6138687.csv
@ -0,0 +1,3 @@
+Type,Name,Request Count,Failure Count,Median Response Time,Average Response Time,Min Response Time,Max Response Time,Average Content Size,Requests/s,Failures/s,50%,66%,75%,80%,90%,95%,98%,99%,99.9%,99.99%,100%
+grpc,/dcache.DcacheService/Write,96465,0,600.0,530.5241670541676,3.931416000114041,2860.153126999876,130.11822940963043,732.274667601832,0.0,600,720,830,880,1100,1200,1300,1500,2300,2900,2900
+,Aggregated,96465,0,600.0,530.5241670541676,3.931416000114041,2860.153126999876,130.11822940963043,732.274667601832,0.0,600,720,830,880,1100,1200,1300,1500,2300,2900,2900
--- a/bench/results/v2/grpc-no-conn-pool-single/nojson/exceptions_1703752024.3059587.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/nojson/exceptions_1703752024.3059587.csv
@ -0,0 +1 @@
+Count,Message,Traceback,Nodes
--- a/bench/results/v2/grpc-no-conn-pool-single/nojson/failures_1703752021.1710336.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/nojson/failures_1703752021.1710336.csv
@ -0,0 +1 @@
+Method,Name,Error,Occurrences
--- a/bench/results/v2/grpc-no-conn-pool-single/nojson/report_1703752029.4987965.html
+++ b/bench/results/v2/grpc-no-conn-pool-single/nojson/report_1703752029.4987965.html
--- a/bench/results/v2/grpc-no-conn-pool-single/nojson/requests_1703752009.4098575.csv
+++ b/bench/results/v2/grpc-no-conn-pool-single/nojson/requests_1703752009.4098575.csv
@ -0,0 +1,3 @@
+Type,Name,Request Count,Failure Count,Median Response Time,Average Response Time,Min Response Time,Max Response Time,Average Content Size,Requests/s,Failures/s,50%,66%,75%,80%,90%,95%,98%,99%,99.9%,99.99%,100%
+grpc,/dcache.DcacheService/AddVisitor,358924,0,79,77.86313645947614,3.354386999944836,123.28810700000759,0.0,4842.970815301002,0.0,79,84,86,88,92,96,100,100,110,120,120
+,Aggregated,358924,0,79,77.86313645947614,3.354386999944836,123.28810700000759,0.0,4842.970815301002,0.0,79,84,86,88,92,96,100,100,110,120,120
--- a/bench/results/v2/grpc-no-conn-pool-single/out.html
+++ b/bench/results/v2/grpc-no-conn-pool-single/out.html
--- a/build.rs
+++ b/build.rs
@ -0,0 +1,10 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    //    tonic_build::configure()
+    //        .out_dir("protoout")
+    //            .compile(
+    //            &["proto/dcache/dcache.proto"],
+    //            &["proto/dcache"],
+    //        )?;
+    tonic_build::compile_protos("proto/dcache/dcache.proto")?;
+    Ok(())
+}
--- a/check.sh
+++ b/check.sh
@ -0,0 +1,9 @@
+#!/bin/bash
+
+protoc \
+--proto_path=${PWD}/proto \
+--proto_path=${PWD}/bufbuild \
+--go_out=${PWD} \
+--go-grpc_out=${PWD} \
+--validate_out="lang=rust:${PWD}" \
+${PWD}/proto/dcache/dcache.proto
--- a/dcache_py/dcache_pb2.py
+++ b/dcache_py/dcache_pb2.py
@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler.  DO NOT EDIT!
+# source: dcache.proto
+# Protobuf Python Version: 4.25.0
+"""Generated protocol buffer code."""
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import descriptor_pool as _descriptor_pool
+from google.protobuf import symbol_database as _symbol_database
+from google.protobuf.internal import builder as _builder
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+
+
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0c\x64\x63\x61\x63he.proto\x12\x06\x64\x63\x61\x63he\"?\n\x05Level\x12\x1a\n\x11visitor_threshold\x18\xad\x02 \x01(\r\x12\x1a\n\x11\x64ifficulty_factor\x18\xae\x02 \x01(\r\")\n\x07\x44\x65\x66\x65nse\x12\x1e\n\x06levels\x18\x91\x03 \x03(\x0b\x32\r.dcache.Level\"@\n\x08MCaptcha\x12\x11\n\x08\x64uration\x18\xf6\x03 \x01(\x04\x12!\n\x07\x64\x65\x66\x65nse\x18\xf7\x03 \x01(\x0b\x32\x0f.dcache.Defense\"E\n\x11\x41\x64\x64\x43\x61ptchaRequest\x12\x0b\n\x02id\x18\xd9\x04 \x01(\t\x12#\n\x08mcaptcha\x18\xda\x04 \x01(\x0b\x32\x10.dcache.MCaptcha\"9\n\x14RenameCaptchaRequest\x12\r\n\x04name\x18\xbd\x05 \x01(\t\x12\x12\n\trename_to\x18\xbe\x05 \x01(\t\"_\n\x0f\x43\x61\x63hePowRequest\x12\x0f\n\x06string\x18\xa1\x06 \x01(\t\x12\x1a\n\x11\x64ifficulty_factor\x18\xa2\x06 \x01(\r\x12\x11\n\x08\x64uration\x18\xa3\x06 \x01(\x04\x12\x0c\n\x03key\x18\xa4\x06 \x01(\t\"E\n\x12\x43\x61\x63heResultRequest\x12\x0e\n\x05token\x18\xb1\x06 \x01(\t\x12\x0c\n\x03key\x18\xb2\x06 \x01(\t\x12\x11\n\x08\x64uration\x18\xb3\x06 \x01(\x04\",\n\x1a\x44\x65leteCaptchaResultRequest\x12\x0e\n\x05token\x18\xb5\x06 \x01(\t\"\x17\n\tCaptchaID\x12\n\n\x02id\x18\x01 \x01(\t\"\x12\n\x04PoID\x12\n\n\x02id\x18\x01 \x01(\t\"A\n\x10\x41\x64\x64VisitorResult\x12\x11\n\x08\x64uration\x18\x85\x07 \x01(\x04\x12\x1a\n\x11\x64ifficulty_factor\x18\x86\x07 \x01(\r\"S\n\x16OptionAddVisitorResult\x12.\n\x06result\x18\x8f\x07 \x01(\x0b\x32\x18.dcache.AddVisitorResultH\x00\x88\x01\x01\x42\t\n\x07_result\"\x1b\n\x0bRaftRequest\x12\x0c\n\x04\x64\x61ta\x18\x01 \x01(\t\"(\n\tRaftReply\x12\x0c\n\x04\x64\x61ta\x18\x01 \x01(\t\x12\r\n\x05\x65rror\x18\x02 \x01(\t\"#\n\x07Learner\x12\n\n\x02id\x18\x01 \x01(\x04\x12\x0c\n\x04\x61\x64\x64r\x18\x02 \x01(\t\"\xbd\x02\n\rDcacheRequest\x12/\n\naddCaptcha\x18\x01 \x01(\x0b\x32\x19.dcache.AddCaptchaRequestH\x00\x12\'\n\naddVisitor\x18\x02 \x01(\x0b\x32\x11.dcache.CaptchaIDH\x00\x12\x35\n\rrenameCaptcha\x18\x03 \x01(\x0b\x32\x1c.dcache.RenameCaptchaRequestH\x00\x12*\n\rremoveCaptcha\x18\x04 \x01(\x0b\x32\x11.dcache.CaptchaIDH\x00\x12+\n\x08\x63\x61\x63hePow\x18\x05 \x01(\x0b\x32\x17.dcache.CachePowRequestH\x00\x12\x31\n\x0b\x63\x61\x63heResult\x18\x06 \x01(\x0b\x32\x1a.dcache.CacheResultRequestH\x00\x42\x0f\n\rDcacheRequest\"\x8b\x01\n\x0e\x44\x63\x61\x63heResponse\x12\x43\n\x19option_add_visitor_result\x18\x01 \x01(\x0b\x32\x1e.dcache.OptionAddVisitorResultH\x00\x12\"\n\x05other\x18\x02 \x01(\x0b\x32\x11.dcache.RaftReplyH\x00\x42\x10\n\x0e\x44\x63\x61\x63heResponse\"=\n\x12\x44\x63\x61\x63heBatchRequest\x12\'\n\x08requests\x18\x01 \x03(\x0b\x32\x15.dcache.DcacheRequest\"@\n\x13\x44\x63\x61\x63heBatchResponse\x12)\n\tresponses\x18\x01 \x03(\x0b\x32\x16.dcache.DcacheResponse2\x97\x06\n\rDcacheService\x12<\n\nAddCaptcha\x12\x19.dcache.AddCaptchaRequest\x1a\x11.dcache.RaftReply\"\x00\x12\x41\n\nAddVisitor\x12\x11.dcache.CaptchaID\x1a\x1e.dcache.OptionAddVisitorResult\"\x00\x12\x42\n\rRenameCaptcha\x12\x1c.dcache.RenameCaptchaRequest\x1a\x11.dcache.RaftReply\"\x00\x12\x37\n\rRemoveCaptcha\x12\x11.dcache.CaptchaID\x1a\x11.dcache.RaftReply\"\x00\x12\x38\n\x08\x43\x61\x63hePow\x12\x17.dcache.CachePowRequest\x1a\x11.dcache.RaftReply\"\x00\x12>\n\x0b\x43\x61\x63heResult\x12\x1a.dcache.CacheResultRequest\x1a\x11.dcache.RaftReply\"\x00\x12N\n\x11PipelineDcacheOps\x12\x1a.dcache.DcacheBatchRequest\x1a\x1b.dcache.DcacheBatchResponse\"\x00\x12\x32\n\nAddLearner\x12\x0f.dcache.Learner\x1a\x11.dcache.RaftReply\"\x00\x12\x31\n\x05Write\x12\x13.dcache.RaftRequest\x1a\x11.dcache.RaftReply\"\x00\x12\x33\n\x07\x46orward\x12\x13.dcache.RaftRequest\x1a\x11.dcache.RaftReply\"\x00\x12\x37\n\rAppendEntries\x12\x13.dcache.RaftRequest\x1a\x11.dcache.RaftReply\x12\x39\n\x0fInstallSnapshot\x12\x13.dcache.RaftRequest\x1a\x11.dcache.RaftReply\x12.\n\x04vote\x12\x13.dcache.RaftRequest\x1a\x11.dcache.RaftReplyb\x06proto3')
+
+_globals = globals()
+_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
+_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'dcache_pb2', _globals)
+if _descriptor._USE_C_DESCRIPTORS == False:
+  DESCRIPTOR._options = None
+  _globals['_LEVEL']._serialized_start=24
+  _globals['_LEVEL']._serialized_end=87
+  _globals['_DEFENSE']._serialized_start=89
+  _globals['_DEFENSE']._serialized_end=130
+  _globals['_MCAPTCHA']._serialized_start=132
+  _globals['_MCAPTCHA']._serialized_end=196
+  _globals['_ADDCAPTCHAREQUEST']._serialized_start=198
+  _globals['_ADDCAPTCHAREQUEST']._serialized_end=267
+  _globals['_RENAMECAPTCHAREQUEST']._serialized_start=269
+  _globals['_RENAMECAPTCHAREQUEST']._serialized_end=326
+  _globals['_CACHEPOWREQUEST']._serialized_start=328
+  _globals['_CACHEPOWREQUEST']._serialized_end=423
+  _globals['_CACHERESULTREQUEST']._serialized_start=425
+  _globals['_CACHERESULTREQUEST']._serialized_end=494
+  _globals['_DELETECAPTCHARESULTREQUEST']._serialized_start=496
+  _globals['_DELETECAPTCHARESULTREQUEST']._serialized_end=540
+  _globals['_CAPTCHAID']._serialized_start=542
+  _globals['_CAPTCHAID']._serialized_end=565
+  _globals['_POID']._serialized_start=567
+  _globals['_POID']._serialized_end=585
+  _globals['_ADDVISITORRESULT']._serialized_start=587
+  _globals['_ADDVISITORRESULT']._serialized_end=652
+  _globals['_OPTIONADDVISITORRESULT']._serialized_start=654
+  _globals['_OPTIONADDVISITORRESULT']._serialized_end=737
+  _globals['_RAFTREQUEST']._serialized_start=739
+  _globals['_RAFTREQUEST']._serialized_end=766
+  _globals['_RAFTREPLY']._serialized_start=768
+  _globals['_RAFTREPLY']._serialized_end=808
+  _globals['_LEARNER']._serialized_start=810
+  _globals['_LEARNER']._serialized_end=845
+  _globals['_DCACHEREQUEST']._serialized_start=848
+  _globals['_DCACHEREQUEST']._serialized_end=1165
+  _globals['_DCACHERESPONSE']._serialized_start=1168
+  _globals['_DCACHERESPONSE']._serialized_end=1307
+  _globals['_DCACHEBATCHREQUEST']._serialized_start=1309
+  _globals['_DCACHEBATCHREQUEST']._serialized_end=1370
+  _globals['_DCACHEBATCHRESPONSE']._serialized_start=1372
+  _globals['_DCACHEBATCHRESPONSE']._serialized_end=1436
+  _globals['_DCACHESERVICE']._serialized_start=1439
+  _globals['_DCACHESERVICE']._serialized_end=2230
+# @@protoc_insertion_point(module_scope)
--- a/dcache_py/dcache_pb2.pyi
+++ b/dcache_py/dcache_pb2.pyi
@ -0,0 +1,156 @@
+from google.protobuf.internal import containers as _containers
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from typing import ClassVar as _ClassVar, Iterable as _Iterable, Mapping as _Mapping, Optional as _Optional, Union as _Union
+
+DESCRIPTOR: _descriptor.FileDescriptor
+
+class Level(_message.Message):
+    __slots__ = ("visitor_threshold", "difficulty_factor")
+    VISITOR_THRESHOLD_FIELD_NUMBER: _ClassVar[int]
+    DIFFICULTY_FACTOR_FIELD_NUMBER: _ClassVar[int]
+    visitor_threshold: int
+    difficulty_factor: int
+    def __init__(self, visitor_threshold: _Optional[int] = ..., difficulty_factor: _Optional[int] = ...) -> None: ...
+
+class Defense(_message.Message):
+    __slots__ = ("levels",)
+    LEVELS_FIELD_NUMBER: _ClassVar[int]
+    levels: _containers.RepeatedCompositeFieldContainer[Level]
+    def __init__(self, levels: _Optional[_Iterable[_Union[Level, _Mapping]]] = ...) -> None: ...
+
+class MCaptcha(_message.Message):
+    __slots__ = ("duration", "defense")
+    DURATION_FIELD_NUMBER: _ClassVar[int]
+    DEFENSE_FIELD_NUMBER: _ClassVar[int]
+    duration: int
+    defense: Defense
+    def __init__(self, duration: _Optional[int] = ..., defense: _Optional[_Union[Defense, _Mapping]] = ...) -> None: ...
+
+class AddCaptchaRequest(_message.Message):
+    __slots__ = ("id", "mcaptcha")
+    ID_FIELD_NUMBER: _ClassVar[int]
+    MCAPTCHA_FIELD_NUMBER: _ClassVar[int]
+    id: str
+    mcaptcha: MCaptcha
+    def __init__(self, id: _Optional[str] = ..., mcaptcha: _Optional[_Union[MCaptcha, _Mapping]] = ...) -> None: ...
+
+class RenameCaptchaRequest(_message.Message):
+    __slots__ = ("name", "rename_to")
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    RENAME_TO_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    rename_to: str
+    def __init__(self, name: _Optional[str] = ..., rename_to: _Optional[str] = ...) -> None: ...
+
+class CachePowRequest(_message.Message):
+    __slots__ = ("string", "difficulty_factor", "duration", "key")
+    STRING_FIELD_NUMBER: _ClassVar[int]
+    DIFFICULTY_FACTOR_FIELD_NUMBER: _ClassVar[int]
+    DURATION_FIELD_NUMBER: _ClassVar[int]
+    KEY_FIELD_NUMBER: _ClassVar[int]
+    string: str
+    difficulty_factor: int
+    duration: int
+    key: str
+    def __init__(self, string: _Optional[str] = ..., difficulty_factor: _Optional[int] = ..., duration: _Optional[int] = ..., key: _Optional[str] = ...) -> None: ...
+
+class CacheResultRequest(_message.Message):
+    __slots__ = ("token", "key", "duration")
+    TOKEN_FIELD_NUMBER: _ClassVar[int]
+    KEY_FIELD_NUMBER: _ClassVar[int]
+    DURATION_FIELD_NUMBER: _ClassVar[int]
+    token: str
+    key: str
+    duration: int
+    def __init__(self, token: _Optional[str] = ..., key: _Optional[str] = ..., duration: _Optional[int] = ...) -> None: ...
+
+class DeleteCaptchaResultRequest(_message.Message):
+    __slots__ = ("token",)
+    TOKEN_FIELD_NUMBER: _ClassVar[int]
+    token: str
+    def __init__(self, token: _Optional[str] = ...) -> None: ...
+
+class CaptchaID(_message.Message):
+    __slots__ = ("id",)
+    ID_FIELD_NUMBER: _ClassVar[int]
+    id: str
+    def __init__(self, id: _Optional[str] = ...) -> None: ...
+
+class PoID(_message.Message):
+    __slots__ = ("id",)
+    ID_FIELD_NUMBER: _ClassVar[int]
+    id: str
+    def __init__(self, id: _Optional[str] = ...) -> None: ...
+
+class AddVisitorResult(_message.Message):
+    __slots__ = ("duration", "difficulty_factor")
+    DURATION_FIELD_NUMBER: _ClassVar[int]
+    DIFFICULTY_FACTOR_FIELD_NUMBER: _ClassVar[int]
+    duration: int
+    difficulty_factor: int
+    def __init__(self, duration: _Optional[int] = ..., difficulty_factor: _Optional[int] = ...) -> None: ...
+
+class OptionAddVisitorResult(_message.Message):
+    __slots__ = ("result",)
+    RESULT_FIELD_NUMBER: _ClassVar[int]
+    result: AddVisitorResult
+    def __init__(self, result: _Optional[_Union[AddVisitorResult, _Mapping]] = ...) -> None: ...
+
+class RaftRequest(_message.Message):
+    __slots__ = ("data",)
+    DATA_FIELD_NUMBER: _ClassVar[int]
+    data: str
+    def __init__(self, data: _Optional[str] = ...) -> None: ...
+
+class RaftReply(_message.Message):
+    __slots__ = ("data", "error")
+    DATA_FIELD_NUMBER: _ClassVar[int]
+    ERROR_FIELD_NUMBER: _ClassVar[int]
+    data: str
+    error: str
+    def __init__(self, data: _Optional[str] = ..., error: _Optional[str] = ...) -> None: ...
+
+class Learner(_message.Message):
+    __slots__ = ("id", "addr")
+    ID_FIELD_NUMBER: _ClassVar[int]
+    ADDR_FIELD_NUMBER: _ClassVar[int]
+    id: int
+    addr: str
+    def __init__(self, id: _Optional[int] = ..., addr: _Optional[str] = ...) -> None: ...
+
+class DcacheRequest(_message.Message):
+    __slots__ = ("addCaptcha", "addVisitor", "renameCaptcha", "removeCaptcha", "cachePow", "cacheResult")
+    ADDCAPTCHA_FIELD_NUMBER: _ClassVar[int]
+    ADDVISITOR_FIELD_NUMBER: _ClassVar[int]
+    RENAMECAPTCHA_FIELD_NUMBER: _ClassVar[int]
+    REMOVECAPTCHA_FIELD_NUMBER: _ClassVar[int]
+    CACHEPOW_FIELD_NUMBER: _ClassVar[int]
+    CACHERESULT_FIELD_NUMBER: _ClassVar[int]
+    addCaptcha: AddCaptchaRequest
+    addVisitor: CaptchaID
+    renameCaptcha: RenameCaptchaRequest
+    removeCaptcha: CaptchaID
+    cachePow: CachePowRequest
+    cacheResult: CacheResultRequest
+    def __init__(self, addCaptcha: _Optional[_Union[AddCaptchaRequest, _Mapping]] = ..., addVisitor: _Optional[_Union[CaptchaID, _Mapping]] = ..., renameCaptcha: _Optional[_Union[RenameCaptchaRequest, _Mapping]] = ..., removeCaptcha: _Optional[_Union[CaptchaID, _Mapping]] = ..., cachePow: _Optional[_Union[CachePowRequest, _Mapping]] = ..., cacheResult: _Optional[_Union[CacheResultRequest, _Mapping]] = ...) -> None: ...
+
+class DcacheResponse(_message.Message):
+    __slots__ = ("option_add_visitor_result", "other")
+    OPTION_ADD_VISITOR_RESULT_FIELD_NUMBER: _ClassVar[int]
+    OTHER_FIELD_NUMBER: _ClassVar[int]
+    option_add_visitor_result: OptionAddVisitorResult
+    other: RaftReply
+    def __init__(self, option_add_visitor_result: _Optional[_Union[OptionAddVisitorResult, _Mapping]] = ..., other: _Optional[_Union[RaftReply, _Mapping]] = ...) -> None: ...
+
+class DcacheBatchRequest(_message.Message):
+    __slots__ = ("requests",)
+    REQUESTS_FIELD_NUMBER: _ClassVar[int]
+    requests: _containers.RepeatedCompositeFieldContainer[DcacheRequest]
+    def __init__(self, requests: _Optional[_Iterable[_Union[DcacheRequest, _Mapping]]] = ...) -> None: ...
+
+class DcacheBatchResponse(_message.Message):
+    __slots__ = ("responses",)
+    RESPONSES_FIELD_NUMBER: _ClassVar[int]
+    responses: _containers.RepeatedCompositeFieldContainer[DcacheResponse]
+    def __init__(self, responses: _Optional[_Iterable[_Union[DcacheResponse, _Mapping]]] = ...) -> None: ...
--- a/dcache_py/dcache_pb2_grpc.py
+++ b/dcache_py/dcache_pb2_grpc.py
@ -0,0 +1,465 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+
+import dcache_py.dcache_pb2 as dcache__pb2
+
+
+class DcacheServiceStub(object):
+    """Missing associated documentation comment in .proto file."""
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.AddCaptcha = channel.unary_unary(
+                '/dcache.DcacheService/AddCaptcha',
+                request_serializer=dcache__pb2.AddCaptchaRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.AddVisitor = channel.unary_unary(
+                '/dcache.DcacheService/AddVisitor',
+                request_serializer=dcache__pb2.CaptchaID.SerializeToString,
+                response_deserializer=dcache__pb2.OptionAddVisitorResult.FromString,
+                )
+        self.RenameCaptcha = channel.unary_unary(
+                '/dcache.DcacheService/RenameCaptcha',
+                request_serializer=dcache__pb2.RenameCaptchaRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.RemoveCaptcha = channel.unary_unary(
+                '/dcache.DcacheService/RemoveCaptcha',
+                request_serializer=dcache__pb2.CaptchaID.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.CachePow = channel.unary_unary(
+                '/dcache.DcacheService/CachePow',
+                request_serializer=dcache__pb2.CachePowRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.CacheResult = channel.unary_unary(
+                '/dcache.DcacheService/CacheResult',
+                request_serializer=dcache__pb2.CacheResultRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.PipelineDcacheOps = channel.unary_unary(
+                '/dcache.DcacheService/PipelineDcacheOps',
+                request_serializer=dcache__pb2.DcacheBatchRequest.SerializeToString,
+                response_deserializer=dcache__pb2.DcacheBatchResponse.FromString,
+                )
+        self.AddLearner = channel.unary_unary(
+                '/dcache.DcacheService/AddLearner',
+                request_serializer=dcache__pb2.Learner.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.Write = channel.unary_unary(
+                '/dcache.DcacheService/Write',
+                request_serializer=dcache__pb2.RaftRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.Forward = channel.unary_unary(
+                '/dcache.DcacheService/Forward',
+                request_serializer=dcache__pb2.RaftRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.AppendEntries = channel.unary_unary(
+                '/dcache.DcacheService/AppendEntries',
+                request_serializer=dcache__pb2.RaftRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.InstallSnapshot = channel.unary_unary(
+                '/dcache.DcacheService/InstallSnapshot',
+                request_serializer=dcache__pb2.RaftRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+        self.vote = channel.unary_unary(
+                '/dcache.DcacheService/vote',
+                request_serializer=dcache__pb2.RaftRequest.SerializeToString,
+                response_deserializer=dcache__pb2.RaftReply.FromString,
+                )
+
+
+class DcacheServiceServicer(object):
+    """Missing associated documentation comment in .proto file."""
+
+    def AddCaptcha(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def AddVisitor(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def RenameCaptcha(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def RemoveCaptcha(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def CachePow(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def CacheResult(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def PipelineDcacheOps(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def AddLearner(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Write(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Forward(self, request, context):
+        """/ Forward a request to other
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def AppendEntries(self, request, context):
+        """raft RPC
+
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def InstallSnapshot(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def vote(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_DcacheServiceServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+            'AddCaptcha': grpc.unary_unary_rpc_method_handler(
+                    servicer.AddCaptcha,
+                    request_deserializer=dcache__pb2.AddCaptchaRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'AddVisitor': grpc.unary_unary_rpc_method_handler(
+                    servicer.AddVisitor,
+                    request_deserializer=dcache__pb2.CaptchaID.FromString,
+                    response_serializer=dcache__pb2.OptionAddVisitorResult.SerializeToString,
+            ),
+            'RenameCaptcha': grpc.unary_unary_rpc_method_handler(
+                    servicer.RenameCaptcha,
+                    request_deserializer=dcache__pb2.RenameCaptchaRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'RemoveCaptcha': grpc.unary_unary_rpc_method_handler(
+                    servicer.RemoveCaptcha,
+                    request_deserializer=dcache__pb2.CaptchaID.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'CachePow': grpc.unary_unary_rpc_method_handler(
+                    servicer.CachePow,
+                    request_deserializer=dcache__pb2.CachePowRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'CacheResult': grpc.unary_unary_rpc_method_handler(
+                    servicer.CacheResult,
+                    request_deserializer=dcache__pb2.CacheResultRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'PipelineDcacheOps': grpc.unary_unary_rpc_method_handler(
+                    servicer.PipelineDcacheOps,
+                    request_deserializer=dcache__pb2.DcacheBatchRequest.FromString,
+                    response_serializer=dcache__pb2.DcacheBatchResponse.SerializeToString,
+            ),
+            'AddLearner': grpc.unary_unary_rpc_method_handler(
+                    servicer.AddLearner,
+                    request_deserializer=dcache__pb2.Learner.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'Write': grpc.unary_unary_rpc_method_handler(
+                    servicer.Write,
+                    request_deserializer=dcache__pb2.RaftRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'Forward': grpc.unary_unary_rpc_method_handler(
+                    servicer.Forward,
+                    request_deserializer=dcache__pb2.RaftRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'AppendEntries': grpc.unary_unary_rpc_method_handler(
+                    servicer.AppendEntries,
+                    request_deserializer=dcache__pb2.RaftRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'InstallSnapshot': grpc.unary_unary_rpc_method_handler(
+                    servicer.InstallSnapshot,
+                    request_deserializer=dcache__pb2.RaftRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+            'vote': grpc.unary_unary_rpc_method_handler(
+                    servicer.vote,
+                    request_deserializer=dcache__pb2.RaftRequest.FromString,
+                    response_serializer=dcache__pb2.RaftReply.SerializeToString,
+            ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+            'dcache.DcacheService', rpc_method_handlers)
+    server.add_generic_rpc_handlers((generic_handler,))
+
+
+ # This class is part of an EXPERIMENTAL API.
+class DcacheService(object):
+    """Missing associated documentation comment in .proto file."""
+
+    @staticmethod
+    def AddCaptcha(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/AddCaptcha',
+            dcache__pb2.AddCaptchaRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def AddVisitor(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/AddVisitor',
+            dcache__pb2.CaptchaID.SerializeToString,
+            dcache__pb2.OptionAddVisitorResult.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def RenameCaptcha(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/RenameCaptcha',
+            dcache__pb2.RenameCaptchaRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def RemoveCaptcha(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/RemoveCaptcha',
+            dcache__pb2.CaptchaID.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def CachePow(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/CachePow',
+            dcache__pb2.CachePowRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def CacheResult(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/CacheResult',
+            dcache__pb2.CacheResultRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def PipelineDcacheOps(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/PipelineDcacheOps',
+            dcache__pb2.DcacheBatchRequest.SerializeToString,
+            dcache__pb2.DcacheBatchResponse.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def AddLearner(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/AddLearner',
+            dcache__pb2.Learner.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def Write(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/Write',
+            dcache__pb2.RaftRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def Forward(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/Forward',
+            dcache__pb2.RaftRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def AppendEntries(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/AppendEntries',
+            dcache__pb2.RaftRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def InstallSnapshot(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/InstallSnapshot',
+            dcache__pb2.RaftRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def vote(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/dcache.DcacheService/vote',
+            dcache__pb2.RaftRequest.SerializeToString,
+            dcache__pb2.RaftReply.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
--- a/proto/dcache/dcache.proto
+++ b/proto/dcache/dcache.proto
@ -0,0 +1,129 @@
+syntax = "proto3";
+
+package dcache;
+
+message Level {
+    uint32 visitor_threshold = 301;
+    uint32 difficulty_factor= 302;
+}
+
+message Defense {
+    repeated Level levels = 401;
+}
+message MCaptcha {
+	uint64 duration = 502;
+	Defense defense = 503;
+}
+
+message AddCaptchaRequest {
+	string id = 601;
+	MCaptcha mcaptcha = 602;
+}
+
+
+message RenameCaptchaRequest {
+	string name = 701;
+	string rename_to = 702;
+}
+
+
+message CachePowRequest {
+	string string= 801;
+	uint32 difficulty_factor = 802;
+	uint64 duration = 803;
+	string key = 804;
+}
+
+message CacheResultRequest {
+	string token =  817;
+	string key = 818;
+	uint64 duration= 819;
+}
+
+message DeleteCaptchaResultRequest {
+	string token = 821;
+}
+
+message CaptchaID{
+	string id = 1;
+}
+
+message PoID{
+	string id = 1;
+}
+
+
+message AddVisitorResult {
+    uint64 duration = 901;
+    uint32 difficulty_factor = 902;
+}
+
+message OptionAddVisitorResult {
+	optional AddVisitorResult result = 911;
+}
+
+message RaftRequest {
+  string data = 1;
+}
+
+message RaftReply {
+  string data = 1;
+  string error = 2;
+}
+
+message Learner {
+    uint64 id = 1;
+    string addr = 2;
+}
+
+
+message DcacheRequest {
+    oneof DcacheRequest {
+	AddCaptchaRequest addCaptcha = 1;
+	CaptchaID addVisitor = 2;
+	RenameCaptchaRequest renameCaptcha = 3;
+	CaptchaID removeCaptcha = 4;
+	CachePowRequest cachePow = 5;
+	CacheResultRequest cacheResult = 6;
+    }
+}
+
+
+message DcacheResponse {
+    oneof DcacheResponse {
+		OptionAddVisitorResult option_add_visitor_result = 1;
+        RaftReply other = 2;
+	}
+}
+
+message DcacheBatchRequest {
+    repeated DcacheRequest requests = 1;
+}
+
+message DcacheBatchResponse {
+    repeated DcacheResponse responses = 1;
+}
+
+service DcacheService {
+	rpc AddCaptcha(AddCaptchaRequest) returns (RaftReply) {}
+	rpc AddVisitor(CaptchaID) returns (OptionAddVisitorResult) {}
+	rpc RenameCaptcha(RenameCaptchaRequest) returns (RaftReply) {}
+	rpc RemoveCaptcha(CaptchaID) returns (RaftReply) {}
+	rpc CachePow(CachePowRequest) returns (RaftReply) {}
+	rpc CacheResult(CacheResultRequest) returns (RaftReply) {}
+
+	rpc PipelineDcacheOps(DcacheBatchRequest) returns (DcacheBatchResponse) {}
+
+
+	rpc AddLearner(Learner) returns (RaftReply) {}
+	rpc Write(RaftRequest) returns (RaftReply) {}
+
+	/// Forward a request to other
+	rpc Forward(RaftRequest) returns (RaftReply) {}
+
+	// raft RPC
+
+	rpc AppendEntries(RaftRequest) returns (RaftReply);
+	rpc InstallSnapshot(RaftRequest) returns (RaftReply);
+	rpc vote(RaftRequest) returns (RaftReply);
+}
--- a/src/app.rs
+++ b/src/app.rs
@ -17,7 +17,6 @@
 */
 use std::collections::BTreeMap;
 use std::sync::Arc;
-use std::time::Duration;

 use openraft::error::RaftError;
 use openraft::BasicNode;
--- a/src/bin/main.rs
+++ b/src/bin/main.rs
@ -16,11 +16,7 @@
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 use clap::Parser;
-use dcache::network::raft_network_impl::DcacheNetwork;
 use dcache::start_example_raft_node;
-use dcache::store::DcacheStore;
-use dcache::DcacheTypeConfig;
-use openraft::Raft;
 use tracing_subscriber::EnvFilter;

 //pub type DcacheRaft = Raft<DcacheTypeConfig, DcacheNetwork, DcacheStore>;
@ -44,7 +40,7 @@ pub struct Opt {
    pub cluster_size: usize,
 }

-#[actix_web::main]
+#[actix_rt::main]
 async fn main() -> std::io::Result<()> {
    // Setup the logger
    tracing_subscriber::fmt()
--- a/src/lib.rs
+++ b/src/lib.rs
@ -20,27 +20,24 @@
 use std::io::Cursor;
 use std::sync::Arc;

-use actix_web::middleware;
-use actix_web::middleware::Logger;
-use actix_web::web::Data;
-use actix_web::App;
-use actix_web::HttpServer;
 use openraft::storage::Adaptor;
 use openraft::BasicNode;
 use openraft::Config;
 use openraft::Raft;
+use tonic::transport::Server;

 use crate::app::DcacheApp;
-use crate::network::api;
-use crate::network::management;
-use crate::network::raft;
 use crate::network::raft_network_impl::DcacheNetwork;
+use crate::protobuf::dcache::dcache_service_client::DcacheServiceClient;
+use crate::protobuf::dcache::dcache_service_server::DcacheServiceServer;
+use crate::protobuf::dcache::Learner;
 use crate::store::DcacheRequest;
 use crate::store::DcacheResponse;
 use crate::store::DcacheStore;

 pub mod app;
 pub mod network;
+mod protobuf;
 pub mod store;

 pub type DcacheNodeId = u64;
@ -98,7 +95,6 @@ pub async fn start_example_raft_node(
    let store = Arc::new(DcacheStore::new(salt));

    let (log_store, state_machine) = Adaptor::new(store.clone());
-    let client = reqwest::Client::new();

    // Create the network layer that will connect and communicate the raft instances and
    // will be used in conjunction with the store created above.
@ -106,7 +102,7 @@ pub async fn start_example_raft_node(
    let (manager_tx, manager_rx) = tokio::sync::mpsc::channel(1000);
    // let health = Arc::new(crate::network::raft_network_impl::HealthLedger::new(manager_tx));
    // let network = Arc::new(DcacheNetwork::new(health));
-    let network = Arc::new(DcacheNetwork::new(manager_tx, client.clone()));
+    let network = Arc::new(DcacheNetwork::new(manager_tx));

    // Create a local raft instance.
    let raft = Raft::new(
@ -120,67 +116,73 @@ pub async fn start_example_raft_node(
    .unwrap();
    raft.enable_heartbeat(true);
    raft.enable_elect(true);
+
+    let captcha = serde_json::json!({
+    "AddCaptcha": {
+        "id": "test_1",
+        "mcaptcha": {
+            "visitor_threshold": 0,
+            "defense": {
+                "levels": [
+                    {"visitor_threshold": 50, "difficulty_factor": 500},
+                    {"visitor_threshold": 5000, "difficulty_factor": 50000},
+                ],
+                "current_visitor_threshold": 0,
+            },
+            "duration": 30,
+        }}});
+    #[derive(serde::Serialize)]
+    struct X {
+        data: String,
+    }
+    let x = X {
+        data: serde_json::to_string(&captcha).unwrap(),
+    };
+    println!("{}", serde_json::to_string(&x).unwrap());
+
    //    raft.enable_tick(true);

    // Create an application that will store all the instances created above, this will
    // be later used on the actix-web services.
-    let app = Data::new(DcacheApp {
+    let app = DcacheApp {
        id: node_id,
        addr: http_addr.clone(),
        raft,
        store,
        config,
        network,
-    });
+    };
+    let app = Arc::new(app);
+    let dcache_service = protobuf::MyDcacheImpl::new(app.clone());

    if introducer_addr == http_addr {
        app.init().await.unwrap();
    }

    let app_copy = app.clone();
-    // Start the actix-web server.
-    let server = HttpServer::new(move || {
-        App::new()
-            .wrap(Logger::default())
-            .wrap(Logger::new("%a %{User-Agent}i"))
-            .wrap(middleware::Compress::default())
-            .app_data(app.clone())
-            // raft internal RPC
-            .service(raft::append)
-            .service(raft::snapshot)
-            .service(raft::vote)
-            // admin API
-            .service(management::init)
-            .service(management::add_learner)
-            .service(management::change_membership)
-            .service(management::metrics)
-            // application API
-            .service(api::write)
-            .service(api::state)
-            .service(api::read)
-            .service(api::pipeline_read)
-            .service(api::pipeline_write)
-        //            .service(api::consistent_read)
-    });

-    let x = server.bind(&http_addr)?;
+    let svc = DcacheServiceServer::new(dcache_service);
+
+    let x = Server::builder()
+        .add_service(svc)
+        .serve(http_addr.clone().parse().unwrap());
+    let server_fut = tokio::spawn(x);

-    let server_fut = tokio::spawn(x.run());
    tokio::time::sleep(std::time::Duration::new(3, 0)).await;

-    let req: (DcacheNodeId, String) = (node_id, http_addr);
-    let c = reqwest::Client::new();
-    c.post(format!("http://{}/add-learner", introducer_addr))
-        .json(&req)
-        .send()
+    let url = format!("http://{}", introducer_addr);
+    let mut client = DcacheServiceClient::connect(url).await.unwrap();
+    client
+        .add_learner(Learner {
+            id: node_id,
+            addr: http_addr,
+        })
        .await
        .unwrap();
-    //    let health_job = tokio::spawn(DcacheApp::health_job(app_copy));

    let health_metrics_handle =
        crate::network::management::HealthMetrics::spawn(app_copy, 5, manager_rx).await;
-    server_fut.await??;
+    server_fut.await?.unwrap();
    health_metrics_handle.abort();
-    //   health_job.abort();
    Ok(())
 }
--- a/src/network/api.rs
+++ b/src/network/api.rs
@ -1,157 +0,0 @@
-/*
- * mCaptcha - A proof of work based DoS protection system
- * Copyright © 2023 Aravinth Manivannan <realravinth@batsense.net>
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-use actix_web::get;
-use actix_web::post;
-use actix_web::web;
-use actix_web::web::Data;
-use actix_web::Responder;
-use libmcaptcha::cache::messages::{CachedPoWConfig, RetrivePoW, VerifyCaptchaResult};
-use libmcaptcha::master::messages::GetInternalData;
-use serde::Deserialize;
-use serde::Serialize;
-use web::Json;
-
-use crate::app::DcacheApp;
-use crate::store::DcacheRequest;
-
-#[post("/write")]
-pub async fn write(
-    app: Data<DcacheApp>,
-    req: Json<DcacheRequest>,
-) -> actix_web::Result<impl Responder> {
-    let response = app.raft.client_write(req.0).await;
-    Ok(Json(response))
-}
-
-#[get("/state")]
-pub async fn state(app: Data<DcacheApp>) -> actix_web::Result<impl Responder> {
-    let sm = app.store.state_machine.read().await;
-    let resp = sm
-        .data
-        .master
-        .send(GetInternalData)
-        .await
-        .unwrap()
-        .await
-        .unwrap()
-        .unwrap();
-    Ok(Json(resp))
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub enum ReadRequest {
-    RetrivePoW(RetrivePoW),                   //Reader
-    VerifyCaptchaResult(VerifyCaptchaResult), //Reader
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone)]
-pub enum ReadResponse {
-    VerifyCaptchaResult(bool),
-    RetrivePoW(Option<CachedPoWConfig>),
-}
-
-#[post("/read")]
-pub async fn read(
-    app: Data<DcacheApp>,
-    req: Json<ReadRequest>,
-) -> actix_web::Result<impl Responder> {
-    let sm = app.store.state_machine.read().await;
-
-    let req = req.into_inner();
-    let res = match req {
-        ReadRequest::RetrivePoW(msg) => {
-            let cache_res = sm
-                .data
-                .cache
-                .send(msg.clone())
-                .await
-                .unwrap()
-                .await
-                .unwrap()
-                .unwrap();
-            ReadResponse::RetrivePoW(cache_res)
-        }
-        ReadRequest::VerifyCaptchaResult(msg) => {
-            let cache_res = sm
-                .data
-                .cache
-                .send(msg.clone())
-                .await
-                .unwrap()
-                .await
-                .unwrap()
-                .unwrap();
-            ReadResponse::VerifyCaptchaResult(cache_res)
-        }
-    };
-    Ok(Json(res))
-}
-
-#[post("/pipeline/read")]
-pub async fn pipeline_read(
-    app: Data<DcacheApp>,
-    requests: Json<Vec<ReadRequest>>,
-) -> actix_web::Result<impl Responder> {
-    let requests = requests.into_inner();
-    let mut responses = Vec::with_capacity(requests.len());
-    let sm = app.store.state_machine.read().await;
-    for request in requests {
-        let res = match request {
-            ReadRequest::RetrivePoW(msg) => {
-                let cache_res = sm
-                    .data
-                    .cache
-                    .send(msg.clone())
-                    .await
-                    .unwrap()
-                    .await
-                    .unwrap()
-                    .unwrap();
-                ReadResponse::RetrivePoW(cache_res)
-            }
-            ReadRequest::VerifyCaptchaResult(msg) => {
-                let cache_res = sm
-                    .data
-                    .cache
-                    .send(msg.clone())
-                    .await
-                    .unwrap()
-                    .await
-                    .unwrap()
-                    .unwrap();
-                ReadResponse::VerifyCaptchaResult(cache_res)
-            }
-        };
-
-        responses.push(res);
-    }
-    Ok(Json(responses))
-}
-
-#[post("/pipeline/write")]
-pub async fn pipeline_write(
-    app: Data<DcacheApp>,
-    requests: Json<Vec<DcacheRequest>>,
-) -> actix_web::Result<impl Responder> {
-    let mut responses = Vec::with_capacity(requests.len());
-    let mut requests = requests.into_inner();
-    for req in requests.drain(0..) {
-        responses.push(app.raft.client_write(req).await);
-    }
-    Ok(Json(responses))
-}
--- a/src/network/management.rs
+++ b/src/network/management.rs
@ -15,66 +15,15 @@
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
-use std::collections::BTreeMap;
-use std::collections::BTreeSet;
 use std::collections::HashMap;
+use std::sync::Arc;

-use actix_web::get;
-use actix_web::post;
-use actix_web::web;
-use actix_web::web::Data;
-use actix_web::Responder;
-use openraft::error::Infallible;
-use openraft::BasicNode;
-use openraft::RaftMetrics;
-use web::Json;
+//use actix_web::web;
+//use actix_web::web::Data;

 use crate::app::DcacheApp;
 use crate::DcacheNodeId;

-#[post("/add-learner")]
-pub async fn add_learner(
-    app: Data<DcacheApp>,
-    req: Json<(DcacheNodeId, String)>,
-) -> actix_web::Result<impl Responder> {
-    let node_id = req.0 .0;
-    let node = BasicNode {
-        addr: req.0 .1.clone(),
-    };
-    let res = app.raft.add_learner(node_id, node, true).await;
-
-    Ok(Json(res))
-}
-
-#[post("/change-membership")]
-pub async fn change_membership(
-    app: Data<DcacheApp>,
-    req: Json<BTreeSet<DcacheNodeId>>,
-) -> actix_web::Result<impl Responder> {
-    let res = app.raft.change_membership(req.0, false).await;
-    Ok(Json(res))
-}
-
-#[post("/init")]
-pub async fn init(app: Data<DcacheApp>) -> actix_web::Result<impl Responder> {
-    let mut nodes = BTreeMap::new();
-    nodes.insert(
-        app.id,
-        BasicNode {
-            addr: app.addr.clone(),
-        },
-    );
-    let res = app.raft.initialize(nodes).await;
-    Ok(Json(res))
-}
-
-#[get("/metrics")]
-pub async fn metrics(app: Data<DcacheApp>) -> actix_web::Result<impl Responder> {
-    let metrics = app.raft.metrics().borrow().clone();
-    let res: Result<RaftMetrics<DcacheNodeId, BasicNode>, Infallible> = Ok(metrics);
-    Ok(Json(res))
-}
-
 use tokio::sync::mpsc;

 #[derive(Debug)]
@ -87,7 +36,7 @@ pub struct HealthMetrics;

 impl HealthMetrics {
    pub async fn spawn(
-        app: Data<DcacheApp>,
+        app: Arc<DcacheApp>,
        threshold: usize,
        mut rx: mpsc::Receiver<HealthStatus>,
    ) -> tokio::task::JoinHandle<()> {
@ -114,7 +63,7 @@ impl HealthMetrics {
                                    new_nodes.push(*node.0);
                                }

-                                let res =
+                                let _res =
                                    app.raft.change_membership(new_nodes, false).await.unwrap();
                            }
                        } else {
@ -128,20 +77,3 @@ impl HealthMetrics {
        tokio::spawn(fut)
    }
 }
-
-//#[get("/self/remove/{id}")]
-//pub async fn remove_node(app: Data<DcacheApp>, id: web::Path<u64>) -> actix_web::Result<impl Responder> {
-//    let cluster_metrics = app.raft.metrics().borrow().clone();
-//    let remote_id: u64 = 3;
-//    let mut new_nodes: Vec<DcacheNodeId> = Vec::new();
-//    for node in cluster_metrics.membership_config.nodes() {
-//        if *node.0 == remote_id {
-//            continue;
-//        }
-//
-//        new_nodes.push(*node.0);
-//    }
-//
-//    let res = app.raft.change_membership(new_nodes, false).await;
-//    Ok(Json(res))
-//}
--- a/src/network/mod.rs
+++ b/src/network/mod.rs
@ -15,7 +15,7 @@
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
-pub mod api;
+//pub mod api;
 pub mod management;
-pub mod raft;
+//pub mod raft;
 pub mod raft_network_impl;
--- a/src/network/raft.rs
+++ b/src/network/raft.rs
@ -1,58 +0,0 @@
-/*
- * mCaptcha - A proof of work based DoS protection system
- * Copyright © 2023 Aravinth Manivannan <realravinth@batsense.net>
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-use actix_web::post;
-use actix_web::web;
-use actix_web::web::Data;
-use actix_web::Responder;
-use openraft::raft::AppendEntriesRequest;
-use openraft::raft::InstallSnapshotRequest;
-use openraft::raft::VoteRequest;
-use web::Json;
-
-use crate::app::DcacheApp;
-use crate::DcacheNodeId;
-use crate::DcacheTypeConfig;
-
-// --- Raft communication
-
-#[post("/raft-vote")]
-pub async fn vote(
-    app: Data<DcacheApp>,
-    req: Json<VoteRequest<DcacheNodeId>>,
-) -> actix_web::Result<impl Responder> {
-    let res = app.raft.vote(req.0).await;
-    Ok(Json(res))
-}
-
-#[post("/raft-append")]
-pub async fn append(
-    app: Data<DcacheApp>,
-    req: Json<AppendEntriesRequest<DcacheTypeConfig>>,
-) -> actix_web::Result<impl Responder> {
-    let res = app.raft.append_entries(req.0).await;
-    Ok(Json(res))
-}
-
-#[post("/raft-snapshot")]
-pub async fn snapshot(
-    app: Data<DcacheApp>,
-    req: Json<InstallSnapshotRequest<DcacheTypeConfig>>,
-) -> actix_web::Result<impl Responder> {
-    let res = app.raft.install_snapshot(req.0).await;
-    Ok(Json(res))
-}
--- a/src/network/raft_network_impl.rs
+++ b/src/network/raft_network_impl.rs
@ -1,6 +1,3 @@
-use std::collections::BTreeMap;
-use std::collections::BTreeSet;
-use std::collections::HashSet;
 /*
 * mCaptcha - A proof of work based DoS protection system
 * Copyright © 2023 Aravinth Manivannan <realravinth@batsense.net>
@ -18,18 +15,13 @@ use std::collections::HashSet;
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
-use std::collections::HashMap;
 use std::sync::Arc;
-use std::sync::RwLock;
-use std::time::Duration;
-use std::time::Instant;

 use async_trait::async_trait;
 use openraft::error::InstallSnapshotError;
 use openraft::error::NetworkError;
 use openraft::error::RPCError;
 use openraft::error::RaftError;
-use openraft::error::RemoteError;
 use openraft::raft::AppendEntriesRequest;
 use openraft::raft::AppendEntriesResponse;
 use openraft::raft::InstallSnapshotRequest;
@ -39,7 +31,6 @@ use openraft::raft::VoteResponse;
 use openraft::BasicNode;
 use openraft::RaftNetwork;
 use openraft::RaftNetworkFactory;
-use reqwest::Client;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
 use tokio::sync::mpsc::Sender;
@ -48,22 +39,30 @@ use super::management::HealthStatus;
 use crate::DcacheNodeId;
 use crate::DcacheTypeConfig;

+use crate::protobuf::dcache::dcache_service_client::DcacheServiceClient;
+use crate::protobuf::dcache::RaftRequest;
+
 #[derive(Clone)]
 pub struct DcacheNetwork {
    pub signal: Sender<HealthStatus>,
-    pub client: Client,
+}
+
+pub enum RPCType {
+    Vote,
+    Snapshot,
+    Append,
 }

 impl DcacheNetwork {
-    pub fn new(signal: Sender<HealthStatus>, client: Client) -> Self {
-        Self { signal, client }
+    pub fn new(signal: Sender<HealthStatus>) -> Self {
+        Self { signal }
    }
    pub async fn send_rpc<Req, Resp, Err>(
        &self,
        target: DcacheNodeId,
        target_node: &BasicNode,
-        uri: &str,
        req: Req,
+        event: RPCType,
    ) -> Result<Resp, RPCError<DcacheNodeId, BasicNode, Err>>
    where
        Req: Serialize,
@ -72,34 +71,51 @@ impl DcacheNetwork {
    {
        let addr = &target_node.addr;

-        let url = format!("http://{}/{}", addr, uri);
+        let url = format!("http://{}", addr);

-        tracing::debug!("send_rpc to url: {}", url);
+        let mut client = DcacheServiceClient::connect(url).await.unwrap();

-        let resp = match self.client.post(url).json(&req).send().await {
-            Ok(resp) => Ok(resp),
+        let res = match event {
+            RPCType::Vote => {
+                client
+                    .vote(RaftRequest {
+                        data: serde_json::to_string(&req).unwrap(),
+                    })
+                    .await
+            }
+
+            RPCType::Snapshot => {
+                client
+                    .install_snapshot(RaftRequest {
+                        data: serde_json::to_string(&req).unwrap(),
+                    })
+                    .await
+            }
+
+            RPCType::Append => {
+                client
+                    .append_entries(RaftRequest {
+                        data: serde_json::to_string(&req).unwrap(),
+                    })
+                    .await
+            }
+        };
+
+        match res {
+            Ok(res) => {
+                let signal2 = self.signal.clone();
+                let fut = async move {
+                    let _ = signal2.send(HealthStatus::Healthy(target)).await;
+                };
+                tokio::spawn(fut);
+                let res = res.into_inner();
+                Ok(serde_json::from_str(&res.data).unwrap())
+            }
            Err(e) => {
-                self.signal.send(HealthStatus::Down(target)).await;
+                let _ = self.signal.send(HealthStatus::Down(target)).await;
                Err(RPCError::Network(NetworkError::new(&e)))
            }
-        }?;
-
-        tracing::debug!("client.post() is sent");
-
-        let res: Result<Resp, Err> = resp
-            .json()
-            .await
-            .map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
-
-        let res = res.map_err(|e| RPCError::RemoteError(RemoteError::new(target, e)));
-        if res.is_ok() {
-            let signal2 = self.signal.clone();
-            let fut = async move {
-                let _ = signal2.send(HealthStatus::Healthy(target)).await;
-            };
-            tokio::spawn(fut);
        }
-        res
    }
 }

@ -134,7 +150,7 @@ impl RaftNetwork<DcacheTypeConfig> for DcacheNetworkConnection {
        RPCError<DcacheNodeId, BasicNode, RaftError<DcacheNodeId>>,
    > {
        self.owner
-            .send_rpc(self.target, &self.target_node, "raft-append", req)
+            .send_rpc(self.target, &self.target_node, req, RPCType::Append)
            .await
    }

@ -146,7 +162,7 @@ impl RaftNetwork<DcacheTypeConfig> for DcacheNetworkConnection {
        RPCError<DcacheNodeId, BasicNode, RaftError<DcacheNodeId, InstallSnapshotError>>,
    > {
        self.owner
-            .send_rpc(self.target, &self.target_node, "raft-snapshot", req)
+            .send_rpc(self.target, &self.target_node, req, RPCType::Append)
            .await
    }

@ -158,7 +174,7 @@ impl RaftNetwork<DcacheTypeConfig> for DcacheNetworkConnection {
        RPCError<DcacheNodeId, BasicNode, RaftError<DcacheNodeId>>,
    > {
        self.owner
-            .send_rpc(self.target, &self.target_node, "raft-vote", req)
+            .send_rpc(self.target, &self.target_node, req, RPCType::Vote)
            .await
    }
 }
--- a/src/protobuf.rs
+++ b/src/protobuf.rs
@ -0,0 +1,402 @@
+use std::sync::Arc;
+
+use libmcaptcha::cache::messages as CacheMessages;
+use libmcaptcha::defense;
+use libmcaptcha::master::messages as MasterMessages;
+use libmcaptcha::mcaptcha;
+use openraft::BasicNode;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use tonic::Response;
+
+use dcache::dcache_request::DcacheRequest as PipelineReq;
+use dcache::dcache_response::DcacheResponse as InnerPipelineRes;
+use dcache::dcache_service_server::DcacheService;
+use dcache::DcacheResponse as OuterPipelineRes;
+use dcache::{Learner, RaftReply, RaftRequest};
+
+use crate::app::DcacheApp;
+use crate::store::{DcacheRequest, DcacheResponse};
+
+pub mod dcache {
+    tonic::include_proto!("dcache"); // The string specified here must match the proto package name
+}
+
+#[derive(Clone)]
+pub struct MyDcacheImpl {
+    app: Arc<DcacheApp>,
+}
+
+impl MyDcacheImpl {
+    pub fn new(app: Arc<DcacheApp>) -> Self {
+        Self { app }
+    }
+}
+
+#[tonic::async_trait]
+impl DcacheService for MyDcacheImpl {
+    async fn add_learner(
+        &self,
+        request: tonic::Request<Learner>,
+    ) -> std::result::Result<tonic::Response<RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let node_id = req.id;
+        let node = BasicNode {
+            addr: req.addr.clone(),
+        };
+        println!("Learner added: {:?}", &req.addr);
+        let res = self.app.raft.add_learner(node_id, node, true).await;
+        Ok(Response::new(res.into()))
+    }
+
+    async fn add_captcha(
+        &self,
+        request: tonic::Request<dcache::AddCaptchaRequest>,
+    ) -> std::result::Result<tonic::Response<RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let res = self
+            .app
+            .raft
+            .client_write(DcacheRequest::AddCaptcha(req.into()))
+            .await;
+        Ok(Response::new(res.into()))
+    }
+
+    async fn add_visitor(
+        &self,
+        request: tonic::Request<dcache::CaptchaId>,
+    ) -> std::result::Result<tonic::Response<dcache::OptionAddVisitorResult>, tonic::Status> {
+        let req = request.into_inner();
+        let res = self
+            .app
+            .raft
+            .client_write(DcacheRequest::AddVisitor(MasterMessages::AddVisitor(
+                req.id,
+            )))
+            .await
+            .map_err(|e| {
+                tonic::Status::new(tonic::Code::Internal, serde_json::to_string(&e).unwrap())
+            })?;
+        match res.data {
+            DcacheResponse::AddVisitorResult(res) => {
+                Ok(Response::new(dcache::OptionAddVisitorResult {
+                    result: res.map(|f| f.into()),
+                }))
+            }
+
+            _ => unimplemented!(),
+        }
+    }
+
+    async fn rename_captcha(
+        &self,
+        request: tonic::Request<dcache::RenameCaptchaRequest>,
+    ) -> std::result::Result<tonic::Response<dcache::RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let res = self
+            .app
+            .raft
+            .client_write(DcacheRequest::RenameCaptcha(req.into()))
+            .await;
+        Ok(Response::new(res.into()))
+    }
+
+    async fn remove_captcha(
+        &self,
+        request: tonic::Request<dcache::CaptchaId>,
+    ) -> std::result::Result<tonic::Response<dcache::RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let res = self
+            .app
+            .raft
+            .client_write(DcacheRequest::RemoveCaptcha(MasterMessages::RemoveCaptcha(
+                req.id,
+            )))
+            .await;
+        Ok(Response::new(res.into()))
+    }
+
+    async fn cache_pow(
+        &self,
+        request: tonic::Request<dcache::CachePowRequest>,
+    ) -> std::result::Result<tonic::Response<dcache::RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let res = self
+            .app
+            .raft
+            .client_write(DcacheRequest::CachePoW(req.into()))
+            .await;
+        Ok(Response::new(res.into()))
+    }
+
+    async fn cache_result(
+        &self,
+        request: tonic::Request<dcache::CacheResultRequest>,
+    ) -> std::result::Result<tonic::Response<dcache::RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let res = self
+            .app
+            .raft
+            .client_write(DcacheRequest::CacheResult(req.into()))
+            .await;
+        Ok(Response::new(res.into()))
+    }
+
+    //    type PipelineDcacheOpsStream =
+    //        Pin<Box<dyn Stream<Item = Result<OuterPipelineRes, tonic::Status>> + Send + 'static>>;
+
+    //    async fn pipeline_dcache_ops(
+    //        &self,
+    //        request: tonic::Request<tonic::Streaming<dcache::DcacheRequest>>,
+    //    ) -> std::result::Result<tonic::Response<Self::PipelineDcacheOpsStream>, tonic::Status> {
+
+    async fn pipeline_dcache_ops(
+        &self,
+        request: tonic::Request<dcache::DcacheBatchRequest>,
+    ) -> Result<Response<dcache::DcacheBatchResponse>, tonic::Status> {
+        let mut reqs = request.into_inner();
+        let mut responses = Vec::with_capacity(reqs.requests.len());
+        for req in reqs.requests.drain(0..) {
+            let res = match req.dcache_request.unwrap() {
+                PipelineReq::AddCaptcha(add_captcha_req) => {
+                    let res = self
+                        .app
+                        .raft
+                        .client_write(DcacheRequest::AddCaptcha(add_captcha_req.into()))
+                        .await;
+                    OuterPipelineRes {
+                        dcache_response: Some(InnerPipelineRes::Other(res.into())),
+                    }
+                }
+                PipelineReq::AddVisitor(add_visitor_req) => {
+                    let res = self
+                        .app
+                        .raft
+                        .client_write(DcacheRequest::AddVisitor(MasterMessages::AddVisitor(
+                            add_visitor_req.id,
+                        )))
+                        .await;
+                    match res {
+                        Err(_) => OuterPipelineRes {
+                            dcache_response: None,
+                        },
+                        Ok(res) => match res.data {
+                            DcacheResponse::AddVisitorResult(res) => {
+                                let res = dcache::OptionAddVisitorResult {
+                                    result: res.map(|f| f.into()),
+                                };
+                                OuterPipelineRes {
+                                    dcache_response: Some(
+                                        InnerPipelineRes::OptionAddVisitorResult(res),
+                                    ),
+                                }
+                            }
+
+                            _ => unimplemented!(),
+                        },
+                    }
+
+                }
+                PipelineReq::RenameCaptcha(rename_captcha_req) => {
+                    let res = self
+                        .app
+                        .raft
+                        .client_write(DcacheRequest::RenameCaptcha(rename_captcha_req.into()))
+                        .await;
+                    OuterPipelineRes {
+                        dcache_response: Some(InnerPipelineRes::Other(res.into())),
+                    }
+                }
+                PipelineReq::RemoveCaptcha(remove_captcha_req) => {
+                    let res = self
+                        .app
+                        .raft
+                        .client_write(DcacheRequest::RemoveCaptcha(MasterMessages::RemoveCaptcha(
+                            remove_captcha_req.id,
+                        )))
+                        .await;
+                    OuterPipelineRes {
+                        dcache_response: Some(InnerPipelineRes::Other(res.into())),
+                    }
+                }
+                PipelineReq::CachePow(cache_pow_req) => {
+                    let res = self
+                        .app
+                        .raft
+                        .client_write(DcacheRequest::CachePoW(cache_pow_req.into()))
+                        .await;
+                    OuterPipelineRes {
+                        dcache_response: Some(InnerPipelineRes::Other(res.into())),
+                    }
+                }
+                PipelineReq::CacheResult(cache_result_req) => {
+                    let res = self
+                        .app
+                        .raft
+                        .client_write(DcacheRequest::CacheResult(cache_result_req.into()))
+                        .await;
+                    OuterPipelineRes {
+                        dcache_response: Some(InnerPipelineRes::Other(res.into())),
+                    }
+                }
+            };
+            responses.push(res);
+        }
+        Ok(Response::new(dcache::DcacheBatchResponse { responses }))
+    }
+
+    async fn write(
+        &self,
+        request: tonic::Request<RaftRequest>,
+    ) -> std::result::Result<tonic::Response<RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let req = serde_json::from_str(&req.data).unwrap();
+        let res = self.app.raft.client_write(req).await;
+        Ok(Response::new(res.into()))
+    }
+    /// / Forward a request to other
+    async fn forward(
+        &self,
+        _request: tonic::Request<RaftRequest>,
+    ) -> std::result::Result<tonic::Response<RaftReply>, tonic::Status> {
+        unimplemented!();
+    }
+    async fn append_entries(
+        &self,
+        request: tonic::Request<RaftRequest>,
+    ) -> std::result::Result<tonic::Response<RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let req = serde_json::from_str(&req.data).unwrap();
+        let res = self.app.raft.append_entries(req).await;
+        Ok(Response::new(res.into()))
+    }
+    async fn install_snapshot(
+        &self,
+        request: tonic::Request<RaftRequest>,
+    ) -> std::result::Result<tonic::Response<RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let req = serde_json::from_str(&req.data).unwrap();
+        let res = self.app.raft.install_snapshot(req).await;
+        Ok(Response::new(res.into()))
+    }
+    async fn vote(
+        &self,
+        request: tonic::Request<RaftRequest>,
+    ) -> std::result::Result<tonic::Response<RaftReply>, tonic::Status> {
+        let req = request.into_inner();
+        let req = serde_json::from_str(&req.data).unwrap();
+        let res = self.app.raft.vote(req).await;
+        Ok(Response::new(res.into()))
+    }
+}
+
+impl<T, E> From<RaftReply> for Result<T, E>
+where
+    T: DeserializeOwned,
+    E: DeserializeOwned,
+{
+    fn from(msg: RaftReply) -> Self {
+        if !msg.data.is_empty() {
+            let resp: T = serde_json::from_str(&msg.data).expect("fail to deserialize");
+            Ok(resp)
+        } else {
+            let err: E = serde_json::from_str(&msg.error).expect("fail to deserialize");
+            Err(err)
+        }
+    }
+}
+
+impl<T, E> From<Result<T, E>> for RaftReply
+where
+    T: Serialize,
+    E: Serialize,
+{
+    fn from(r: Result<T, E>) -> Self {
+        match r {
+            Ok(x) => {
+                let data = serde_json::to_string(&x).expect("fail to serialize");
+                RaftReply {
+                    data,
+                    error: Default::default(),
+                }
+            }
+            Err(e) => {
+                let error = serde_json::to_string(&e).expect("fail to serialize");
+                RaftReply {
+                    data: Default::default(),
+                    error,
+                }
+            }
+        }
+    }
+}
+
+impl From<dcache::AddCaptchaRequest> for MasterMessages::AddSite {
+    fn from(value: dcache::AddCaptchaRequest) -> Self {
+        let req_mcaptcha = value.mcaptcha.unwrap();
+        let mut defense = req_mcaptcha.defense.unwrap();
+        let mut new_defense = defense::DefenseBuilder::default();
+        for level in defense.levels.drain(0..) {
+            new_defense
+                .add_level(
+                    defense::LevelBuilder::default()
+                        .difficulty_factor(level.difficulty_factor)
+                        .unwrap()
+                        .visitor_threshold(level.visitor_threshold)
+                        .build()
+                        .unwrap(),
+                )
+                .unwrap();
+        }
+        let defense = new_defense.build().unwrap();
+        let mcaptcha = mcaptcha::MCaptchaBuilder::default()
+            .defense(defense)
+            .duration(req_mcaptcha.duration)
+            .build()
+            .unwrap();
+
+        Self {
+            id: value.id,
+            mcaptcha,
+        }
+    }
+}
+
+impl From<libmcaptcha::master::AddVisitorResult> for dcache::AddVisitorResult {
+    fn from(value: libmcaptcha::master::AddVisitorResult) -> Self {
+        Self {
+            duration: value.duration,
+            difficulty_factor: value.difficulty_factor,
+        }
+    }
+}
+
+impl From<dcache::RenameCaptchaRequest> for MasterMessages::Rename {
+    fn from(value: dcache::RenameCaptchaRequest) -> Self {
+        Self {
+            name: value.name,
+            rename_to: value.rename_to,
+        }
+    }
+}
+
+impl From<dcache::CachePowRequest> for CacheMessages::CachePoW {
+    fn from(value: dcache::CachePowRequest) -> Self {
+        Self {
+            string: value.string,
+            difficulty_factor: value.difficulty_factor,
+            duration: value.duration,
+            key: value.key,
+        }
+    }
+}
+
+impl From<dcache::CacheResultRequest> for CacheMessages::CacheResult {
+    fn from(value: dcache::CacheResultRequest) -> Self {
+        Self {
+            token: value.token,
+            key: value.key,
+            duration: value.duration,
+        }
+    }
+}
--- a/src/store/mod.rs
+++ b/src/store/mod.rs
@ -23,7 +23,6 @@ use std::ops::RangeBounds;
 use std::sync::Arc;
 use std::sync::Mutex;

-use libmcaptcha::cache::messages::CachedPoWConfig;
 use libmcaptcha::AddVisitorResult;
 use libmcaptcha::MCaptcha;
 use openraft::async_trait::async_trait;
@ -48,15 +47,11 @@ use openraft::Vote;
 use serde::Deserialize;
 use serde::Serialize;
 use tokio::sync::RwLock;
-use url::quirks::set_pathname;

 use crate::DcacheNodeId;
 use crate::DcacheTypeConfig;

-use actix::prelude::*;
-use libmcaptcha::cache::messages::{
-    CachePoW, CacheResult, DeleteCaptchaResult, DeletePoW, RetrivePoW, VerifyCaptchaResult,
-};
+use libmcaptcha::cache::messages::{CachePoW, CacheResult, DeleteCaptchaResult, DeletePoW};
 use libmcaptcha::master::messages::{
    AddSite as AddCaptcha, AddVisitor, GetInternalData, RemoveCaptcha, Rename as RenameCaptcha,
    SetInternalData,
@ -124,7 +119,7 @@ impl PersistableStateMachine {
            .unwrap()
            .unwrap();
        Self {
-            last_applied_log: m.last_applied_log.clone(),
+            last_applied_log: m.last_applied_log,
            last_membership: m.last_membership.clone(),
            data: internal_data,
        }
--- a/test.py
+++ b/test.py
@ -17,6 +17,14 @@

 from pprint import pprint
 import requests
+import grpc
+import json
+
+from dcache_py import dcache_pb2 as dcache
+from dcache_py.dcache_pb2 import RaftRequest
+from dcache_py.dcache_pb2_grpc import DcacheServiceStub
+
+# import dcache_py.dcache_resources


 def init(host: str):
@ -82,12 +90,13 @@ host = "localhost:9001"
 peers = [(2, "localhost:9002"), (3, "localhost:9003"), (4, "localhost:9004")]
 captcha_id = "test_1"

+
 def initialize_cluster():
    init(host)
    for peer_id, peer in peers:
        add_host(host=host, id=peer_id, peer=peer)

-    switch_to_cluster(host, nodes=[1, 2,3,4])
+    switch_to_cluster(host, nodes=[1, 2, 3, 4])

    add_captcha(host, captcha_id)
    add_vote(host, captcha_id)
@ -95,7 +104,75 @@ def initialize_cluster():
        add_vote(host, captcha_id)


+def grpc_add_vote(stub: DcacheServiceStub, captcha_id: str):
+    msg = dcache.CaptchaID(id=captcha_id)
+    #    msg = RaftRequest(data=json.dumps({"AddVisitor": captcha_id}))
+    #    resp = stub.Write(msg)
+    resp = stub.AddVisitor(msg)
+    pprint(resp)
+
+
+def grpc_add_captcha(stub: DcacheServiceStub, captcha_id: str):
+    msg = dcache.AddCaptchaRequest(
+        id=captcha_id,
+        mcaptcha=dcache.MCaptcha(
+            duration=30,
+            defense=dcache.Defense(
+                levels=[
+                    dcache.Level(visitor_threshold=50, difficulty_factor=500),
+                    dcache.Level(visitor_threshold=5000, difficulty_factor=50000),
+                ]
+            ),
+        ),
+    )
+    #    params = {
+    #        "AddCaptcha": {
+    #            "id": captcha_id,
+    #            "mcaptcha": {
+    #                "defense": {
+    #                    "levels": [
+    #                        {"visitor_threshold": 50, "difficulty_factor": 500},
+    #                        {"visitor_threshold": 5000, "difficulty_factor": 50000},
+    #                    ],
+    #                    "current_visitor_threshold": 0,
+    #                },
+    #                "duration": 30,
+    #            },
+    #        }
+    #    }
+    #    msg = RaftRequest(data = json.dumps(params))
+    resp = stub.AddCaptcha(msg)
+    pprint(f"Captcha added {captcha_id}: {resp}")
+
+
+
+
+msgs = []
+for _ in range(0,1000):
+    msgs.append(
+    dcache.DcacheRequest(addVisitor=dcache.CaptchaID(id=captcha_id)),
+    )
+
+msgs = dcache.DcacheBatchRequest(requests=msgs)
+
+
+
+def grpc_pipeline_add_vote(stub):
+   responses = stub.PipelineDcacheOps(msgs)
+   for r in responses.responses:
+       print(f"received respo: {r}")
+
+
+
+
+def grpc_run():
+    with grpc.insecure_channel(host) as channel:
+        stub = DcacheServiceStub(channel)
+        grpc_add_captcha(stub, captcha_id)
+        grpc_pipeline_add_vote(stub)
+
+        #grpc_add_vote(stub, captcha_id)

 if __name__ == "__main__":
-
-    add_vote("localhost:9002", captcha_id)
+    grpc_run()
+    # add_vote("localhost:9002", captcha_id)