Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use Call method of rpctype Package

Best Syzkaller code snippet using rpctype.Call

rpc.go

Source:rpc.go

1package meta2import (3	"encoding/binary"4	"errors"5	"fmt"6	"io"7	"io/ioutil"8	"log"9	"net"10	"time"11	"github.com/gogo/protobuf/proto"12	"github.com/hashicorp/raft"13	"github.com/influxdb/influxdb/meta/internal"14)15// Max size of a message before we treat the size as invalid16const (17	MaxMessageSize    = 1024 * 1024 * 102418	leaderDialTimeout = 10 * time.Second19)20// rpc handles request/response style messaging between cluster nodes21type rpc struct {22	logger         *log.Logger23	tracingEnabled bool24	store interface {25		cachedData() *Data26		enableLocalRaft() error27		IsLeader() bool28		Leader() string29		Peers() ([]string, error)30		SetPeers(addrs []string) error31		AddPeer(host string) error32		CreateNode(host string) (*NodeInfo, error)33		NodeByHost(host string) (*NodeInfo, error)34		WaitForDataChanged() error35	}36}37// JoinResult defines the join result structure.38type JoinResult struct {39	RaftEnabled bool40	RaftNodes   []string41	NodeID      uint6442}43// Reply defines the interface for Reply objects.44type Reply interface {45	GetHeader() *internal.ResponseHeader46}47// proxyLeader proxies the connection to the current raft leader48func (r *rpc) proxyLeader(conn *net.TCPConn, buf []byte) {49	if r.store.Leader() == "" {50		r.sendError(conn, "no leader detected during proxyLeader")51		return52	}53	leaderConn, err := net.DialTimeout("tcp", r.store.Leader(), leaderDialTimeout)54	if err != nil {55		r.sendError(conn, fmt.Sprintf("dial leader: %v", err))56		return57	}58	defer leaderConn.Close()59	leaderConn.Write([]byte{MuxRPCHeader})60	// re-write the original message to the leader61	leaderConn.Write(buf)62	if err := proxy(leaderConn.(*net.TCPConn), conn); err != nil {63		r.sendError(conn, fmt.Sprintf("leader proxy error: %v", err))64	}65}66// handleRPCConn reads a command from the connection and executes it.67func (r *rpc) handleRPCConn(conn net.Conn) {68	defer conn.Close()69	// RPC connections should execute on the leader.  If we are not the leader,70	// proxy the connection to the leader so that clients an connect to any node71	// in the cluster.72	r.traceCluster("rpc connection from: %v", conn.RemoteAddr())73	// Read and execute request.74	typ, buf, err := r.readMessage(conn)75	// Handle unexpected RPC errors76	if err != nil {77		r.sendError(conn, err.Error())78		return79	}80	if !r.store.IsLeader() && typ != internal.RPCType_PromoteRaft {81		r.proxyLeader(conn.(*net.TCPConn), pack(typ, buf))82		return83	}84	typ, resp, err := r.executeMessage(conn, typ, buf)85	// Handle unexpected RPC errors86	if err != nil {87		r.sendError(conn, err.Error())88		return89	}90	// Set the status header and error message91	if reply, ok := resp.(Reply); ok {92		reply.GetHeader().OK = proto.Bool(err == nil)93		if err != nil {94			reply.GetHeader().Error = proto.String(err.Error())95		}96	}97	r.sendResponse(conn, typ, resp)98}99func (r *rpc) readMessage(conn net.Conn) (internal.RPCType, []byte, error) {100	// Read request size.101	var sz uint64102	if err := binary.Read(conn, binary.BigEndian, &sz); err != nil {103		return internal.RPCType_Error, nil, fmt.Errorf("read size: %s", err)104	}105	if sz == 0 {106		return internal.RPCType_Error, nil, fmt.Errorf("invalid message size: %d", sz)107	}108	if sz >= MaxMessageSize {109		return internal.RPCType_Error, nil, fmt.Errorf("max message size of %d exceeded: %d", MaxMessageSize, sz)110	}111	// Read request.112	buf := make([]byte, sz)113	if _, err := io.ReadFull(conn, buf); err != nil {114		return internal.RPCType_Error, nil, fmt.Errorf("read request: %s", err)115	}116	// Determine the RPC type117	rpcType := internal.RPCType(btou64(buf[0:8]))118	buf = buf[8:]119	r.traceCluster("recv %v request on: %v", rpcType, conn.RemoteAddr())120	return rpcType, buf, nil121}122func (r *rpc) executeMessage(conn net.Conn, rpcType internal.RPCType, buf []byte) (internal.RPCType, proto.Message, error) {123	switch rpcType {124	case internal.RPCType_FetchData:125		var req internal.FetchDataRequest126		if err := proto.Unmarshal(buf, &req); err != nil {127			return internal.RPCType_Error, nil, fmt.Errorf("fetch request unmarshal: %v", err)128		}129		resp, err := r.handleFetchData(&req)130		return rpcType, resp, err131	case internal.RPCType_Join:132		var req internal.JoinRequest133		if err := proto.Unmarshal(buf, &req); err != nil {134			return internal.RPCType_Error, nil, fmt.Errorf("join request unmarshal: %v", err)135		}136		resp, err := r.handleJoinRequest(&req)137		return rpcType, resp, err138	case internal.RPCType_PromoteRaft:139		var req internal.PromoteRaftRequest140		if err := proto.Unmarshal(buf, &req); err != nil {141			return internal.RPCType_Error, nil, fmt.Errorf("promote to raft request unmarshal: %v", err)142		}143		resp, err := r.handlePromoteRaftRequest(&req)144		return rpcType, resp, err145	default:146		return internal.RPCType_Error, nil, fmt.Errorf("unknown rpc type:%v", rpcType)147	}148}149func (r *rpc) sendResponse(conn net.Conn, typ internal.RPCType, resp proto.Message) {150	// Marshal the response back to a protobuf151	buf, err := proto.Marshal(resp)152	if err != nil {153		r.logger.Printf("unable to marshal response: %v", err)154		return155	}156	// Encode response back to connection.157	if _, err := conn.Write(pack(typ, buf)); err != nil {158		r.logger.Printf("unable to write rpc response: %s", err)159	}160}161func (r *rpc) sendError(conn net.Conn, msg string) {162	r.traceCluster(msg)163	resp := &internal.ErrorResponse{164		Header: &internal.ResponseHeader{165			OK:    proto.Bool(false),166			Error: proto.String(msg),167		},168	}169	r.sendResponse(conn, internal.RPCType_Error, resp)170}171// handleFetchData handles a request for the current nodes meta data172func (r *rpc) handleFetchData(req *internal.FetchDataRequest) (*internal.FetchDataResponse, error) {173	var (174		b    []byte175		data *Data176		err  error177	)178	for {179		data = r.store.cachedData()180		if data.Index != req.GetIndex() {181			b, err = data.MarshalBinary()182			if err != nil {183				return nil, err184			}185			break186		}187		if !req.GetBlocking() {188			break189		}190		if err := r.store.WaitForDataChanged(); err != nil {191			return nil, err192		}193	}194	return &internal.FetchDataResponse{195		Header: &internal.ResponseHeader{196			OK: proto.Bool(true),197		},198		Index: proto.Uint64(data.Index),199		Term:  proto.Uint64(data.Term),200		Data:  b}, nil201}202// handleJoinRequest handles a request to join the cluster203func (r *rpc) handleJoinRequest(req *internal.JoinRequest) (*internal.JoinResponse, error) {204	r.traceCluster("join request from: %v", *req.Addr)205	node, err := func() (*NodeInfo, error) {206		// attempt to create the node207		node, err := r.store.CreateNode(*req.Addr)208		// if it exists, return the existing node209		if err == ErrNodeExists {210			node, err = r.store.NodeByHost(*req.Addr)211			if err != nil {212				return node, err213			}214			r.logger.Printf("existing node re-joined: id=%v addr=%v", node.ID, node.Host)215		} else if err != nil {216			return nil, fmt.Errorf("create node: %v", err)217		}218		peers, err := r.store.Peers()219		if err != nil {220			return nil, fmt.Errorf("list peers: %v", err)221		}222		// If we have less than 3 nodes, add them as raft peers if they are not223		// already a peer224		if len(peers) < MaxRaftNodes && !raft.PeerContained(peers, *req.Addr) {225			r.logger.Printf("adding new raft peer: nodeId=%v addr=%v", node.ID, *req.Addr)226			if err = r.store.AddPeer(*req.Addr); err != nil {227				return node, fmt.Errorf("add peer: %v", err)228			}229		}230		return node, err231	}()232	nodeID := uint64(0)233	if node != nil {234		nodeID = node.ID235	}236	if err != nil {237		return nil, err238	}239	// get the current raft peers240	peers, err := r.store.Peers()241	if err != nil {242		return nil, fmt.Errorf("list peers: %v", err)243	}244	return &internal.JoinResponse{245		Header: &internal.ResponseHeader{246			OK: proto.Bool(true),247		},248		EnableRaft: proto.Bool(raft.PeerContained(peers, *req.Addr)),249		RaftNodes:  peers,250		NodeID:     proto.Uint64(nodeID),251	}, err252}253func (r *rpc) handlePromoteRaftRequest(req *internal.PromoteRaftRequest) (*internal.PromoteRaftResponse, error) {254	r.traceCluster("promote raft request from: %v", *req.Addr)255	// Need to set the local store peers to match what we are about to join256	if err := r.store.SetPeers(req.RaftNodes); err != nil {257		return nil, err258	}259	if err := r.store.enableLocalRaft(); err != nil {260		return nil, err261	}262	if !contains(req.RaftNodes, *req.Addr) {263		req.RaftNodes = append(req.RaftNodes, *req.Addr)264	}265	if err := r.store.SetPeers(req.RaftNodes); err != nil {266		return nil, err267	}268	return &internal.PromoteRaftResponse{269		Header: &internal.ResponseHeader{270			OK: proto.Bool(true),271		},272		Success: proto.Bool(true),273	}, nil274}275// pack returns a TLV style byte slice encoding the size of the payload, the RPC type276// and the RPC data277func pack(typ internal.RPCType, b []byte) []byte {278	buf := u64tob(uint64(len(b)) + 8)279	buf = append(buf, u64tob(uint64(typ))...)280	buf = append(buf, b...)281	return buf282}283// fetchMetaData returns the latest copy of the meta store data from the current284// leader.285func (r *rpc) fetchMetaData(blocking bool) (*Data, error) {286	assert(r.store != nil, "store is nil")287	// Retrieve the current known leader.288	leader := r.store.Leader()289	if leader == "" {290		return nil, errors.New("no leader detected during fetchMetaData")291	}292	var index, term uint64293	data := r.store.cachedData()294	if data != nil {295		index = data.Index296		term = data.Index297	}298	resp, err := r.call(leader, &internal.FetchDataRequest{299		Index:    proto.Uint64(index),300		Term:     proto.Uint64(term),301		Blocking: proto.Bool(blocking),302	})303	if err != nil {304		return nil, err305	}306	switch t := resp.(type) {307	case *internal.FetchDataResponse:308		// If data is nil, then the term and index we sent matches the leader309		if t.GetData() == nil {310			return nil, nil311		}312		ms := &Data{}313		if err := ms.UnmarshalBinary(t.GetData()); err != nil {314			return nil, fmt.Errorf("rpc unmarshal metadata: %v", err)315		}316		return ms, nil317	case *internal.ErrorResponse:318		return nil, fmt.Errorf("rpc failed: %s", t.GetHeader().GetError())319	default:320		return nil, fmt.Errorf("rpc failed: unknown response type: %v", t.String())321	}322}323// join attempts to join a cluster at remoteAddr using localAddr as the current324// node's cluster address325func (r *rpc) join(localAddr, remoteAddr string) (*JoinResult, error) {326	req := &internal.JoinRequest{327		Addr: proto.String(localAddr),328	}329	resp, err := r.call(remoteAddr, req)330	if err != nil {331		return nil, err332	}333	switch t := resp.(type) {334	case *internal.JoinResponse:335		return &JoinResult{336			RaftEnabled: t.GetEnableRaft(),337			RaftNodes:   t.GetRaftNodes(),338			NodeID:      t.GetNodeID(),339		}, nil340	case *internal.ErrorResponse:341		return nil, fmt.Errorf("rpc failed: %s", t.GetHeader().GetError())342	default:343		return nil, fmt.Errorf("rpc failed: unknown response type: %v", t.String())344	}345}346// enableRaft attempts to promote a node at remoteAddr using localAddr as the current347// node's cluster address348func (r *rpc) enableRaft(addr string, peers []string) error {349	req := &internal.PromoteRaftRequest{350		Addr:      proto.String(addr),351		RaftNodes: peers,352	}353	resp, err := r.call(addr, req)354	if err != nil {355		return err356	}357	switch t := resp.(type) {358	case *internal.PromoteRaftResponse:359		return nil360	case *internal.ErrorResponse:361		return fmt.Errorf("rpc failed: %s", t.GetHeader().GetError())362	default:363		return fmt.Errorf("rpc failed: unknown response type: %v", t.String())364	}365}366// call sends an encoded request to the remote leader and returns367// an encoded response value.368func (r *rpc) call(dest string, req proto.Message) (proto.Message, error) {369	// Determine type of request370	var rpcType internal.RPCType371	switch t := req.(type) {372	case *internal.JoinRequest:373		rpcType = internal.RPCType_Join374	case *internal.FetchDataRequest:375		rpcType = internal.RPCType_FetchData376	case *internal.PromoteRaftRequest:377		rpcType = internal.RPCType_PromoteRaft378	default:379		return nil, fmt.Errorf("unknown rpc request type: %v", t)380	}381	// Create a connection to the leader.382	conn, err := net.DialTimeout("tcp", dest, leaderDialTimeout)383	if err != nil {384		return nil, fmt.Errorf("rpc dial: %v", err)385	}386	defer conn.Close()387	// Write a marker byte for rpc messages.388	_, err = conn.Write([]byte{MuxRPCHeader})389	if err != nil {390		return nil, err391	}392	b, err := proto.Marshal(req)393	if err != nil {394		return nil, fmt.Errorf("rpc marshal: %v", err)395	}396	// Write request size & bytes.397	if _, err := conn.Write(pack(rpcType, b)); err != nil {398		return nil, fmt.Errorf("write %v rpc: %s", rpcType, err)399	}400	data, err := ioutil.ReadAll(conn)401	if err != nil {402		return nil, fmt.Errorf("read %v rpc: %v", rpcType, err)403	}404	// Should always have a size and type405	if exp := 16; len(data) < exp {406		r.traceCluster("recv: %v", string(data))407		return nil, fmt.Errorf("rpc %v failed: short read: got %v, exp %v", rpcType, len(data), exp)408	}409	sz := btou64(data[0:8])410	if len(data[8:]) != int(sz) {411		r.traceCluster("recv: %v", string(data))412		return nil, fmt.Errorf("rpc %v failed: short read: got %v, exp %v", rpcType, len(data[8:]), sz)413	}414	// See what response type we got back, could get a general error response415	rpcType = internal.RPCType(btou64(data[8:16]))416	data = data[16:]417	var resp proto.Message418	switch rpcType {419	case internal.RPCType_Join:420		resp = &internal.JoinResponse{}421	case internal.RPCType_FetchData:422		resp = &internal.FetchDataResponse{}423	case internal.RPCType_Error:424		resp = &internal.ErrorResponse{}425	case internal.RPCType_PromoteRaft:426		resp = &internal.PromoteRaftResponse{}427	default:428		return nil, fmt.Errorf("unknown rpc response type: %v", rpcType)429	}430	if err := proto.Unmarshal(data, resp); err != nil {431		return nil, fmt.Errorf("rpc unmarshal: %v", err)432	}433	if reply, ok := resp.(Reply); ok {434		if !reply.GetHeader().GetOK() {435			return nil, fmt.Errorf("rpc %v failed: %s", rpcType, reply.GetHeader().GetError())436		}437	}438	return resp, nil439}440func (r *rpc) traceCluster(msg string, args ...interface{}) {441	if r.tracingEnabled {442		r.logger.Printf("rpc: "+msg, args...)443	}444}445func u64tob(v uint64) []byte {446	b := make([]byte, 8)447	binary.BigEndian.PutUint64(b, v)448	return b449}450func btou64(b []byte) uint64 {451	return binary.BigEndian.Uint64(b)452}453func contains(s []string, e string) bool {454	for _, a := range s {455		if a == e {456			return true457		}458	}459	return false460}...

Call

Using AI Code Generation

1import (2type Args struct {3}4type Quotient struct {5}6func main() {7 client, err := rpc.DialHTTP("tcp", "

Call

Using AI Code Generation

1import (2func main() {3    client, err := rpc.DialHTTP("tcp", "localhost:1234")4    if err != nil {5        fmt.Println(err)6    }7    err = client.Call("HelloService.Hello", "World", &reply)8    if err != nil {9        fmt.Println(err)10    }11    fmt.Println(reply)12}

Call

Using AI Code Generation

1import (2type Args struct {3}4func main() {5	client, err := rpc.DialHTTP("tcp", "localhost:1234")6	if err != nil {7		log.Fatal("dialing:", err)8	}9	args := Args{17, 8}10	err = client.Call("Arith.Multiply", args, &reply)11	if err != nil {12		log.Fatal("arith error:", err)13	}14	fmt.Printf("Arith: %d*%d=%d15}16import (17type Args struct {18}19func (t *Arith) Multiply(args *Args, reply *int) error {20}21func main() {22	arith := new(Arith)23	rpc.Register(arith)24	rpc.HandleHTTP()25	l, e := net.Listen("tcp", ":1234")26	if e != nil {27		log.Fatal("listen error:", e)28	}29	http.Serve(l, nil)30}312017/03/02 20:00:37 rpc.Register: method "Arith.Multiply" has 1 input parameters; needs exactly three32import (33type Args struct {34}35func (t *Arith) Multiply(args *Args, reply *int) error {36}37func main() {38	arith := new(Arith)39	rpc.Register(arith)40	rpc.HandleHTTP()41	l, e := net.Listen("tcp", ":1234")42	if e != nil {43		log.Fatal("listen error:", e)44	}45	http.Serve(l, nil)46}47import (

Call

Using AI Code Generation

1import (2type Args struct {3}4type Quotient struct {5}6func (t *Arith) Multiply(args *Args, reply *int) error {7}8func (t *Arith) Divide(args *Args, quo *Quotient) error {9    if args.B == 0 {10        return fmt.Errorf("divide by zero")11    }12}13func main() {14    arith := new(Arith)15    rpc.Register(arith)16    rpc.HandleHTTP()17    l, e := net.Listen("tcp", ":1234")18    if e != nil {19        log.Fatal("listen error:", e)20    }21    go http.Serve(l, nil)22    client, err := rpc.DialHTTP("tcp", "

Call

Using AI Code Generation

1import (2type Args struct {3}4type Quotient struct {5}6func main() {

Call

Using AI Code Generation

1import (2type Args struct {3}4type Quotient struct {5}6func (t *Arith) Multiply(args *Args, reply *int) error {7}8func (t *Arith) Divide(args *Args, quo *Quotient) error {9	if args.B == 0 {10		return fmt.Errorf("divide by zero")11	}12}13func main() {14	arith := new(Arith)15	rpc.Register(arith)16	rpc.HandleHTTP()17	l, e := net.Listen("tcp", ":1234")18	if e != nil {19		fmt.Println("listen error:", e)20	}21	go http.Serve(l, nil)22	client, err := rpc.DialHTTP("tcp", "

Call

Using AI Code Generation

1import (2type Args struct {3}4func main() {5	client, err := rpc.DialHTTP("tcp", "localhost:1234")6	if err != nil {7		fmt.Println(err)8	}9	args := Args{7, 8}10	err = client.Call("Arith.Multiply", args, &reply)11	if err != nil {12		fmt.Println(err)13	}14	fmt.Printf("Arith: %d*%d=%d", args.A, args.B, reply)15}16import (17type Args struct {18}19type Quotient struct {20}21func (t *Arith) Multiply(args *Args, reply *int) error {22}23func (t *Arith) Divide(args *Args, quo *Quotient) error {24	if args.B == 0 {25		return fmt.Errorf("divide by zero")26	}27}28func main() {29	arith := new(Arith)30	rpc.Register(arith)31	tcpAddr, err := net.ResolveTCPAddr("tcp", ":1234")32	if err != nil {33		fmt.Println(err)34	}35	listener, err := net.ListenTCP("tcp", tcpAddr)36	if err != nil {37		fmt.Println(err)38	}39	for {40		conn, err := listener.Accept()41		if err != nil {42		}43		go jsonrpc.ServeConn(conn)44	}45}

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.