<commit_before>package rpc

import (
	"reflect"
	"sync"
	"net/http"
	"io/ioutil"
)

type Server struct {
	sync.Mutex
	
	network string
	addr string
	funcs map[string]reflect.Value

	listener net.Listener
	running  bool
	objs interface{}
}

func NewServer(network, addr string) *Server {
	RegisterType(RpcError{})

	s := new(Server)
	s.network = network
	s.addr = addr

	s.funcs = make(map[string]reflect.Value)
	return s
}

func (s *Server) Run() error {
	var err error
	s.listener, err = net.Listen(s.network, s.addr)
	if err != nil {
		return err
	}

	s.running = true

	for s.running {
		conn, err := s.listener.Accept()
		if err != nil {
			continue
		}
		go s.onConn(conn)
	}

	return nil
}

func (s *Server) RegisterObj(obj interface{}) (err error) {
	typ := reflect.TypeOf(obj)
	for i := 0; i < typ.NumMethod(); i++ {
		method := typ.Method(i)
		mname := method.Name
		err = s.Register(mname, method.Func.Interface())		
		if err != nil {
			return
		}
	}

	s.obj = obj
	return	
}

func (s *Server) register(name string, f interface{}) (err error) {
	defer func() {
		if e := recover(); e != nil {
			err = fmt.Errorf("%s is not callable", name)
		}
	}()

	v := reflect.ValueOf(f)

	//to check f is function
	v.Type().NumIn()

	nOut := v.Type().NumOut()
	if nOut == 0 || v.Type().Out(nOut-1).Kind() != reflect.Interface {
		err = fmt.Errorf("%s return final output param must be error interface", name)
		return
	}

	_, b := v.Type().Out(nOut - 1).MethodByName("Error")
	if !b {
		err = fmt.Errorf("%s return final output param must be error interface", name)
		return
	}

	s.Lock()
	if _, ok := s.funcs[name]; ok {
		err = fmt.Errorf("%s has registered", name)
		s.Unlock()
		return
	}

	s.funcs[name] = v
	s.Unlock()
	return
}

func routeTo(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()
	data, err := ioutil.ReadAll(r.Body)
	if err != nil {
		println("read body error", err.Error())
	}

	config := &Config{}
	err := json.Unmarshal(data, config)
	if err != nil {
		println("unmarshal err", err.Error())
	}

	name := config.Name
	args := config.Args
	

		
	
}

func (s *Server) handle(name string, args []interface{}) ([]byte, error) {
	name, args, err := decodeData(data)
	if err != nil {
		return nil, err
	}

	s.Lock()
	f, ok := s.funcs[name]
	s.Unlock()
	if !ok {
		return nil, fmt.Errorf("rpc %s not registered", name)
	}

	inValues := make([]reflect.Value, len(args) + 1)
	inValues[0] = reflect.ValueOf(s.obj)

	for i := 0; i < len(args); i++ {<commit_msg>fixed server.go<commit_after>package rpc

import (
	"reflect"
	"sync"
	"net/http"
	"io/ioutil"
)

type Server struct {
	mu sync.Mutex
	
	network string
	addr string
	funcs map[string]reflect.Value

	listener net.Listener
	running  bool
	objs interface{}
}

func NewServer(network, addr string) *Server {
	RegisterType(RpcError{})

	s := new(Server)
	s.mu := &sync.Mutex{} 

	s.network = network
	s.addr = addr

	s.funcs = make(map[string]reflect.Value)
	
	return s
}

func (s *Server) Run() error {
	err := s.start()
	if err != nil {
		return err
	}

	return nil
}

func (s *Server) RegisterObj(obj interface{}) (err error) {
	typ := reflect.TypeOf(obj)
	for i := 0; i < typ.NumMethod(); i++ {
		method := typ.Method(i)
		mname := method.Name
		err = s.Register(mname, method.Func.Interface())		
		if err != nil {
			return
		}
	}

	s.obj = obj
	return	
}

func (s *Server) register(name string, f interface{}) (err error) {
	defer func() {
		if e := recover(); e != nil {
			err = fmt.Errorf("%s is not callable", name)
		}
	}()

	v := reflect.ValueOf(f)

	//to check f is function
	v.Type().NumIn()

	nOut := v.Type().NumOut()
	if nOut == 0 || v.Type().Out(nOut-1).Kind() != reflect.Interface {
		err = fmt.Errorf("%s return final output param must be error interface", name)
		return
	}

	_, b := v.Type().Out(nOut - 1).MethodByName("Error")
	if !b {
		err = fmt.Errorf("%s return final output param must be error interface", name)
		return
	}

	mu.Lock()
	if _, ok := s.funcs[name]; ok {
		err = fmt.Errorf("%s has registered", name)
		mu.Unlock()
		return
	}

	s.funcs[name] = v
	mu.Unlock()
	return
}

func (s *Server) start() error {
	http.HandleFunc("/Router", routeTo)
	return http.ListenAndServe(":" + strconv.Itoa(config.Port), nil)
}

func routeTo(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()
	data, err := ioutil.ReadAll(r.Body)
	if err != nil {
		println("read body error", err.Error())
	}
	
	var f = &interface{}
	err := json.Unmarshal(data, f)
	if err != nil {
		println("unmarshal err", err.Error())
	}

	m := f.(map[string]interface{})
	dataMap := make(map[string]interface{})

	for k, v := range m {
		dataMap[k] = v
	}

	str, ok := dataMap[name]
	if !ok {
		println("rpc name %s not exist", name)
	}

	
	if name, ok := na.(string); !ok {
		println("rpc name is not a string")
	}

	if args, ok := na.([]interface{}); !ok {
		println("rpc args is not a array")
	}
	
	
	if _, err := handle(name, args); err != nil {
		println("handle func err", err.Error())
	}
}

func (s *Server) handle(name string, args []interface{}) ([]byte, error) {
	name, args, err := decodeData(data)
	if err != nil {
		return nil, err
	}

	mu.Lock()
	f, ok := s.funcs[name]
	mu.Unlock()
	if !ok {
		return nil, fmt.Errorf("rpc %s not registered", name)
	}

	inValues := make([]reflect.Value, len(args) + 1)
	inValues[0] = reflect.ValueOf(s.obj)

	for i := 0; i < len(args); i++ {
{'fraction_non_alphanumeric': 0.1722149897712479, 'fraction_numerical': 0.002603682350753208, 'mean_word_length': 3.3127506014434642, 'pattern_counts': {'":': 1, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2334597', 'n_tokens_mistral': 2158, 'n_tokens_neox': 2033, 'n_words': 566}