// rpc call routine and receive reply, waiting if necessary
// to can be set to SendToAll to do a broadcast
rpc call routine and receive reply, waiting if necessary
to can be set to SendToAll to do a broadcast
// rpc call routine and receive reply waiting if necessary // to can be set to SendToAll to do a broadcast
func (twoway *Bilateral) Call(to []string, method string, args interface{}, results interface{}, optionalTimeout ...time.Duration) error {
	if nil != to && 0 == len(to) {
		to = nil
	}
	timeout := DEFAULT_CALL_TIMEOUT
	if len(optionalTimeout) == 1 {
		timeout = optionalTimeout[0]
	} else if len(optionalTimeout) == 1 {
		return errors.New("too many timeout values")
	}
	if timeout < MINIMUM_CALL_TIMEOUT {
		return errors.New("timeout is too short")
	}
	select {
	case <-twoway.shutdownAll:
		return errors.New("RPC system shutting down")
	default:
	}
	ptr := reflect.TypeOf(results)
	if reflect.Ptr != ptr.Kind() && !reflect.ValueOf(results).IsNil() && reflect.Slice != ptr.Elem().Kind() {
		return errors.New("must be a non-nil pointer to slice")
	}
	theSlice := ptr.Elem()
	theStruct := theSlice.Elem()
	if reflect.Struct != theStruct.Kind() || 3 != theStruct.NumField() {
		return errors.New("must be a struct with three fields")
	}
	fromField, hasFrom := theStruct.FieldByName("From")
	_, hasReply := theStruct.FieldByName("Reply")
	errField, hasErr := theStruct.FieldByName("Err")
	if !hasFrom || !hasReply || !hasErr {
		return errors.New("must be a struct with From, Reply and Err fields")
	}
	if reflect.String != fromField.Type.Kind() {
		return errors.New("struct From field must be string")
	}
	if reflect.Interface != errField.Type.Kind() {
		return errors.New("struct Err field must be error")
	}
	done := make(chan *rpcClientRequestData)
	log.Infof("send for method: %s", method)
	twoway.rpcClientRequestChannel <- &rpcClientRequestData{
		id:         nonce(),
		to:         to,
		timeout:    timeout,
		method:     method,
		args:       args,
		structType: theStruct,
		done:       done,
		reply:      nil, 
		count:      0,   
	}
	var request *rpcClientRequestData
loop:
	for retry := 0; retry < DEFAULT_RETRIES; retry += 1 {
		log.Infof("wait for result for method: %s", method)
		request = <-done
		log.Debugf("request for method in loop: %s : %v", method, request)
		if nil != request.reply && len(request.reply) > 0 {
			break loop
		}
		done = make(chan *rpcClientRequestData)
		twoway.rpcClientRequestChannel <- &rpcClientRequestData{
			id:         nonce(),
			to:         to,
			timeout:    timeout,
			method:     method,
			args:       args,
			structType: theStruct,
			done:       done,
			reply:      nil, 
			count:      0,   
		}
		log.Infof("timeout method: %s : %v", method, request)
	}
	log.Debugf("request for method: %s : %v", method, request)
	if nil == request.reply || len(request.reply) == 0 {
		return errors.New("no results")
	}
	received := reflect.MakeSlice(theSlice, 0, len(to))
	for r := range request.reply {
		received = reflect.Append(received, r.Elem())
	}
	reflect.ValueOf(results).Elem().Set(received)
	log.Debugf("Result for method: %s : %v", method, results)
	return nil
}
func ( twoway * Bilateral ) Call ( to [ ] string , method string , args interface { } , results interface { } , optionalTimeout ... time . Duration ) error { if nil != to && 0 == len ( to ) { to = nil } timeout := DEFAULT_CALL_TIMEOUT if len ( optionalTimeout ) == 1 { timeout = optionalTimeout [ 0 ] } else if len ( optionalTimeout ) == 1 { return errors . New ( " " ) } if timeout < MINIMUM_CALL_TIMEOUT { return errors . New ( " " ) } select { case <- twoway . shutdownAll : return errors . New ( " " ) default : } ptr := reflect . TypeOf ( results ) if reflect . Ptr != ptr . Kind ( ) && ! reflect . ValueOf ( results ) . IsNil ( ) && reflect . Slice != ptr . Elem ( ) . Kind ( ) { return errors . New ( " " ) } theSlice := ptr . Elem ( ) theStruct := theSlice . Elem ( ) if reflect . Struct != theStruct . Kind ( ) || 3 != theStruct . NumField ( ) { return errors . New ( " " ) } fromField , hasFrom := theStruct . FieldByName ( " " ) _ , hasReply := theStruct . FieldByName ( " " ) errField , hasErr := theStruct . FieldByName ( " " ) if ! hasFrom || ! hasReply || ! hasErr { return errors . New ( " " ) } if reflect . String != fromField . Type . Kind ( ) { return errors . New ( " " ) } if reflect . Interface != errField . Type . Kind ( ) { return errors . New ( " " ) } done := make ( chan * rpcClientRequestData ) log . Infof ( " " , method ) twoway . rpcClientRequestChannel <- & rpcClientRequestData { id : nonce ( ) , to : to , timeout : timeout , method : method , args : args , structType : theStruct , done : done , reply : nil , count : 0 , } var request * rpcClientRequestData loop : for retry := 0 ; retry < DEFAULT_RETRIES ; retry += 1 { log . Infof ( " " , method ) request = <- done log . Debugf ( " " , method , request ) if nil != request . reply && len ( request . reply ) > 0 { break loop } done = make ( chan * rpcClientRequestData ) twoway . rpcClientRequestChannel <- & rpcClientRequestData { id : nonce ( ) , to : to , timeout : timeout , method : method , args : args , structType : theStruct , done : done , reply : nil , count : 0 , } log . Infof ( " " , method , request ) } log . Debugf ( " " , method , request ) if nil == request . reply || len ( request . reply ) == 0 { return errors . New ( " " ) } received := reflect . MakeSlice ( theSlice , 0 , len ( to ) ) for r := range request . reply { received = reflect . Append ( received , r . Elem ( ) ) } reflect . ValueOf ( results ) . Elem ( ) . Set ( received ) log . Debugf ( " " , method , results ) return nil }
rpc call routine and receive reply, waiting if necessary
to can be set to SendToAll to do a broadcast
rpc call routine and receive reply waiting if necessary to can be set to SendToAll to do a broadcast
// ptr to slice of 's' - some kind of struct // added later // added later // do retries // the processed request will be returned here // added later // added later