Skip to content

Commit fc3cd81

Browse files
qinzuoyanqinzuoyan
committed
add multi_server_sample
1 parent 50f53c6 commit fc3cd81

6 files changed

Lines changed: 404 additions & 0 deletions

File tree

‎sample/multi_server_sample/Makefile‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
# Copyright (c) 2014 Baidu.com, Inc. All rights reserved.
2+
# Use of this source code is governed by a BSD-style license that can be
3+
# found in the LICENSE file. See the AUTHORS file for names of contributors.
4+
5+
#-----------------------------------------------
6+
## Sofa-pbrpc path containing `include'and `lib'.
7+
##
8+
## Check file exist:
9+
## $(SOFA_PBRPC)/include/sofa/pbrpc/pbrpc.h
10+
## $(SOFA_PBRPC)/lib/libsofa-pbrpc.a
11+
##
12+
SOFA_PBRPC=../../output
13+
#-----------------------------------------------
14+
15+
#-----------------------------------------------
16+
# Uncomment exactly one of the lines labelled (A), (B), and (C) below
17+
# to switch between compilation modes.
18+
#
19+
OPT ?= -O2 # (A) Production use (optimized mode)
20+
# OPT ?= -g2 # (B) Debug mode, w/ full line-level debugging symbols
21+
# OPT ?= -O2 -g2 # (C) Profiling mode: opt, but w/debugging symbols
22+
#-----------------------------------------------
23+
24+
#-----------------------------------------------
25+
# !!! Do not change the following lines !!!
26+
#-----------------------------------------------
27+
28+
include ../../depends.mk
29+
30+
CXX=g++
31+
INCPATH=-I. -I$(SOFA_PBRPC)/include -I$(BOOST_HEADER_DIR) -I$(PROTOBUF_DIR)/include \
32+
-I$(SNAPPY_DIR)/include -I$(ZLIB_DIR)/include
33+
CXXFLAGS += $(OPT) -pipe -W -Wall -fPIC -D_GNU_SOURCE -D__STDC_LIMIT_MACROS $(INCPATH)
34+
35+
LIBRARY=$(SOFA_PBRPC)/lib/libsofa-pbrpc.a $(PROTOBUF_DIR)/lib/libprotobuf.a $(SNAPPY_DIR)/lib/libsnappy.a
36+
LDFLAGS += -L$(ZLIB_DIR)/lib -lpthread -lrt -lz
37+
38+
PROTO_SRC=echo_service.proto
39+
PROTO_OBJ=$(patsubst %.proto,%.pb.o,$(PROTO_SRC))
40+
PROTO_OPTIONS=--proto_path=. --proto_path=$(SOFA_PBRPC)/include --proto_path=$(PROTOBUF_DIR)/include
41+
42+
BIN=server client
43+
44+
all: check_depends $(BIN)
45+
46+
.PHONY: check_depends clean
47+
48+
check_depends:
49+
@if [ ! -f "$(PROTOBUF_DIR)/include/google/protobuf/message.h" ]; then echo "ERROR: need protobuf header"; exit 1; fi
50+
@if [ ! -f "$(PROTOBUF_DIR)/lib/libprotobuf.a" ]; then echo "ERROR: need protobuf lib"; exit 1; fi
51+
@if [ ! -f "$(PROTOBUF_DIR)/bin/protoc" ]; then echo "ERROR: need protoc binary"; exit 1; fi
52+
@if [ ! -f "$(SNAPPY_DIR)/include/snappy.h" ]; then echo "ERROR: need snappy header"; exit 1; fi
53+
@if [ ! -f "$(SNAPPY_DIR)/lib/libsnappy.a" ]; then echo "ERROR: need snappy lib"; exit 1; fi
54+
@if [ ! -f "$(SOFA_PBRPC)/include/sofa/pbrpc/pbrpc.h" ]; then echo "ERROR: need sofa-pbrpc header"; exit 1; fi
55+
@if [ ! -f "$(SOFA_PBRPC)/lib/libsofa-pbrpc.a" ]; then echo "ERROR: need sofa-pbrpc lib"; exit 1; fi
56+
57+
clean:
58+
@rm -f $(BIN) *.o *.pb.*
59+
60+
rebuild: clean all
61+
62+
server: $(PROTO_OBJ) server.o
63+
$(CXX) $^ -o $@ $(LIBRARY) $(LDFLAGS)
64+
65+
client: $(PROTO_OBJ) client.o
66+
$(CXX) $^ -o $@ $(LIBRARY) $(LDFLAGS)
67+
68+
%.pb.o: %.pb.cc
69+
$(CXX) $(CXXFLAGS) -c $< -o $@
70+
71+
%.pb.cc: %.proto
72+
$(PROTOBUF_DIR)/bin/protoc $(PROTO_OPTIONS) --cpp_out=. $<
73+
74+
%.o: %.cc $(PROTO_OBJ)
75+
$(CXX) $(CXXFLAGS) -c $< -o $@
76+

‎sample/multi_server_sample/README‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
1, make
2+
3+
2, start servers:
4+
./server 127.0.0.1 12345 &>/dev/null &
5+
./server 127.0.0.1 12346 &>/dev/null &
6+
./server 127.0.0.1 12347 &>/dev/null &
7+
8+
3, start client:
9+
./client address_list.txt
10+
11+
4, remove one address from `address_list.txt'.
12+
13+
5, signal client to reload address list from `address_list.txt':
14+
killall -s SIGTERM client
15+
16+
6, add a new address into `address_list.txt'.
17+
18+
7, signal client to reload address list from `address_list.txt':
19+
killall -s SIGTERM client
20+
21+
8, test done, stop all servers:
22+
killall server
23+

‎sample/multi_server_sample/address_list.txt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
127.0.0.1:12345
2+
127.0.0.1:12346
3+
127.0.0.1:12347

‎sample/multi_server_sample/client.cc‎

Lines changed: 197 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,197 @@
1+
// Copyright (c) 2014 Baidu.com, Inc. All rights reserved.
2+
// Use of this source code is governed by a BSD-style license that can be
3+
// found in the LICENSE file.
4+
//
5+
// Author: qinzuoyan01@baidu.com (Qin Zuoyan)
6+
7+
#include <stdio.h>
8+
#include <signal.h>
9+
#include <unistd.h>
10+
#include <fstream>
11+
#include <vector>
12+
#include <set>
13+
14+
#include <sofa/pbrpc/pbrpc.h>
15+
#include "echo_service.pb.h"
16+
17+
class ReloadableAddressProvider : public sofa::pbrpc::RpcChannel::AddressProvider
18+
{
19+
public:
20+
typedef std::vector<sofa::pbrpc::RpcChannel::EventHandler*> EventHandlerVector;
21+
public:
22+
ReloadableAddressProvider(const std::string& addr_file) : _addr_file(addr_file) {
23+
Reload();
24+
}
25+
26+
virtual ~ReloadableAddressProvider() {
27+
sofa::pbrpc::ScopedLocker<sofa::pbrpc::MutexLock> _(_lock);
28+
for (EventHandlerVector::iterator it = _handler_list.begin();
29+
it != _handler_list.end(); ++it) {
30+
delete *it;
31+
}
32+
}
33+
34+
virtual void GetInitAddress(std::vector<std::string>* address_list) {
35+
sofa::pbrpc::ScopedLocker<sofa::pbrpc::MutexLock> _(_lock);
36+
address_list->assign(_addr_set.begin(), _addr_set.end());
37+
}
38+
39+
virtual bool RegisterEventHandler(sofa::pbrpc::RpcChannel::EventHandler* event_handler) {
40+
sofa::pbrpc::ScopedLocker<sofa::pbrpc::MutexLock> _(_lock);
41+
_handler_list.push_back(event_handler);
42+
return true;
43+
}
44+
45+
void Reload() {
46+
sofa::pbrpc::ScopedLocker<sofa::pbrpc::MutexLock> _(_lock);
47+
SLOG(NOTICE, "start reloading address list from file \"%s\"", _addr_file.c_str());
48+
// open file
49+
std::ifstream ifs(_addr_file.c_str(), std::ifstream::in);
50+
if (!ifs.good()) {
51+
SLOG(ERROR, "open address list file \"%s\" fail", _addr_file.c_str());
52+
return;
53+
}
54+
// read new addresses
55+
std::set<std::string> new_addr_set;
56+
std::string addr;
57+
while (std::getline(ifs, addr)) {
58+
if (!addr.empty()) {
59+
new_addr_set.insert(addr);
60+
}
61+
}
62+
// make diff
63+
std::vector<std::string> add_list;
64+
std::vector<std::string> remove_list;
65+
std::set<std::string>::iterator old_it = _addr_set.begin();
66+
std::set<std::string>::iterator old_end = _addr_set.end();
67+
std::set<std::string>::iterator new_it = new_addr_set.begin();
68+
std::set<std::string>::iterator new_end = new_addr_set.end();
69+
while (old_it != old_end && new_it != new_end) {
70+
if (*old_it == *new_it) {
71+
// keep
72+
++old_it;
73+
++new_it;
74+
}
75+
else if (*old_it < *new_it) {
76+
// remove
77+
remove_list.push_back(*old_it);
78+
++old_it;
79+
}
80+
else {
81+
// add
82+
add_list.push_back(*new_it);
83+
++new_it;
84+
}
85+
}
86+
if (old_it != old_end) {
87+
remove_list.insert(remove_list.end(), old_it, old_end);
88+
}
89+
if (new_it != new_end) {
90+
add_list.insert(add_list.end(), new_it, new_end);
91+
}
92+
// notice handler
93+
if (!add_list.empty() || !remove_list.empty()) {
94+
for (EventHandlerVector::iterator it = _handler_list.begin();
95+
it != _handler_list.end(); ++it) {
96+
if (!add_list.empty()) {
97+
(*it)->OnAddressAdded(add_list);
98+
}
99+
if (!remove_list.empty()) {
100+
(*it)->OnAddressRemoved(remove_list);
101+
}
102+
}
103+
}
104+
// update _addr_set
105+
_addr_set = new_addr_set;
106+
}
107+
108+
private:
109+
std::string _addr_file;
110+
sofa::pbrpc::MutexLock _lock;
111+
std::set<std::string> _addr_set;
112+
EventHandlerVector _handler_list;
113+
};
114+
115+
static ReloadableAddressProvider* g_address_provider;
116+
117+
static void sigcatcher(int sig)
118+
{
119+
SLOG(NOTICE, "signal catched: %d", sig);
120+
if (g_address_provider) {
121+
g_address_provider->Reload();
122+
}
123+
}
124+
125+
int main(int argc, char** argv)
126+
{
127+
SOFA_PBRPC_SET_LOG_LEVEL(NOTICE);
128+
129+
if (argc < 2) {
130+
fprintf(stderr, "USAGE: %s <address-list-file>\n", argv[0]);
131+
return EXIT_FAILURE;
132+
}
133+
134+
std::string addr_file = argv[1];
135+
g_address_provider = new ReloadableAddressProvider(addr_file);
136+
137+
signal(SIGTERM, &sigcatcher);
138+
139+
// Define an rpc client.
140+
sofa::pbrpc::RpcClientOptions client_options;
141+
sofa::pbrpc::RpcClient* rpc_client = new sofa::pbrpc::RpcClient(client_options);
142+
143+
// Define an rpc channel.
144+
sofa::pbrpc::RpcChannelOptions channel_options;
145+
sofa::pbrpc::RpcChannel* rpc_channel =
146+
new sofa::pbrpc::RpcChannel(rpc_client, g_address_provider, channel_options);
147+
148+
// Define an rpc stub.
149+
sofa::pbrpc::test::EchoServer_Stub* stub =
150+
new sofa::pbrpc::test::EchoServer_Stub(rpc_channel);
151+
152+
while (true) {
153+
// Prepare parameters.
154+
sofa::pbrpc::RpcController* cntl = new sofa::pbrpc::RpcController();
155+
cntl->SetTimeout(3000);
156+
sofa::pbrpc::test::EchoRequest* request = new sofa::pbrpc::test::EchoRequest();
157+
request->set_message("Hello from qinzuoyan01");
158+
sofa::pbrpc::test::EchoResponse* response = new sofa::pbrpc::test::EchoResponse();
159+
160+
// Sync call.
161+
stub->Echo(cntl, request, response, NULL);
162+
163+
// Check if the request has been sent.
164+
// If has been sent, then can get the sent bytes.
165+
SLOG(NOTICE, "RemoteAddress=%s", cntl->RemoteAddress().c_str());
166+
SLOG(NOTICE, "IsRequestSent=%s", cntl->IsRequestSent() ? "true" : "false");
167+
if (cntl->IsRequestSent()) {
168+
SLOG(NOTICE, "LocalAddress=%s", cntl->LocalAddress().c_str());
169+
SLOG(NOTICE, "SentBytes=%ld", cntl->SentBytes());
170+
}
171+
172+
// Check if failed.
173+
if (cntl->Failed()) {
174+
SLOG(ERROR, "request failed: %s", cntl->ErrorText().c_str());
175+
}
176+
else {
177+
SLOG(NOTICE, "request succeed: %s", response->message().c_str());
178+
}
179+
180+
// Destroy objects.
181+
delete cntl;
182+
delete request;
183+
delete response;
184+
185+
sleep(1);
186+
}
187+
188+
delete stub;
189+
delete rpc_channel;
190+
delete rpc_client;
191+
delete g_address_provider;
192+
g_address_provider = NULL;
193+
194+
return EXIT_SUCCESS;
195+
}
196+
197+
/* vim: set ts=4 sw=4 sts=4 tw=100 */

‎sample/multi_server_sample/echo_service.proto‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package sofa.pbrpc.test;
2+
3+
option cc_generic_services = true;
4+
option java_generic_services = true;
5+
6+
message EchoRequest {
7+
required string message = 1;
8+
}
9+
10+
message EchoResponse {
11+
required string message = 1;
12+
}
13+
14+
service EchoServer {
15+
rpc Echo(EchoRequest) returns(EchoResponse);
16+
}

0 commit comments

Comments
 (0)