unistack-org/micro
4
0
Fork 3
Code Issues 19 Pull Requests 1 Actions Packages Projects Releases 259 Wiki Activity

make use of micro/go-log

Browse Source
This commit is contained in:
Asim Aslam 2017-05-11 20:43:42 +01:00
parent 39fb03fba6
commit bab586b71e
6 changed files with 31 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
broker/http_broker.go
View File

@ -6,7 +6,6 @@ import (
"fmt" "fmt"
"io" "io"
"io/ioutil" "io/ioutil"
"log"
"math/rand" "math/rand"
"net" "net"
"net/http" "net/http"
@ -17,6 +16,7 @@ import (
"sync" "sync"
"time" "time"
"github.com/micro/go-log"
"github.com/micro/go-micro/broker/codec/json" "github.com/micro/go-micro/broker/codec/json"
"github.com/micro/go-micro/errors" "github.com/micro/go-micro/errors"
"github.com/micro/go-micro/registry" "github.com/micro/go-micro/registry"
@ -249,7 +249,7 @@ func (h *httpBroker) start() error {
return err return err
} }
log.Printf("Broker Listening on %s", l.Addr().String()) log.Logf("Broker Listening on %s", l.Addr().String())
h.address = l.Addr().String() h.address = l.Addr().String()
go http.Serve(l, h.mux) go http.Serve(l, h.mux)

6
selector/cache/cache.go vendored
View File

@ -1,10 +1,10 @@
package cache package cache
import ( import (
"log"
"sync" "sync"
"time" "time"
"github.com/micro/go-log"
"github.com/micro/go-micro/registry" "github.com/micro/go-micro/registry"
"github.com/micro/go-micro/selector" "github.com/micro/go-micro/selector"
) )
@ -241,14 +241,14 @@ func (c *cacheSelector) run() {
// create new watcher // create new watcher
w, err := c.so.Registry.Watch() w, err := c.so.Registry.Watch()
if err != nil { if err != nil {
log.Println(err) log.Log(err)
time.Sleep(time.Second) time.Sleep(time.Second)
continue continue
} }
// watch for events // watch for events
if err := c.watch(w); err != nil { if err := c.watch(w); err != nil {
log.Println(err) log.Log(err)
continue continue
} }
} }

14
server/rpc_server.go
View File

@ -2,13 +2,13 @@ package server
import ( import (
"fmt" "fmt"
"log"
"runtime/debug" "runtime/debug"
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"time" "time"
"github.com/micro/go-log"
"github.com/micro/go-micro/broker" "github.com/micro/go-micro/broker"
"github.com/micro/go-micro/codec" "github.com/micro/go-micro/codec"
"github.com/micro/go-micro/metadata" "github.com/micro/go-micro/metadata"
@ -53,7 +53,7 @@ func (s *rpcServer) accept(sock transport.Socket) {
sock.Close() sock.Close()
if r := recover(); r != nil { if r := recover(); r != nil {
log.Print(r, string(debug.Stack())) log.Log(r, string(debug.Stack()))
} }
}() }()
@ -101,7 +101,7 @@ func (s *rpcServer) accept(sock transport.Socket) {
// TODO: needs better error handling // TODO: needs better error handling
if err := s.rpc.serveRequest(ctx, codec, ct); err != nil { if err := s.rpc.serveRequest(ctx, codec, ct); err != nil {
log.Printf("Unexpected error serving request, closing socket: %v", err) log.Logf("Unexpected error serving request, closing socket: %v", err)
return return
} }
} }
@ -252,7 +252,7 @@ func (s *rpcServer) Register() error {
s.Unlock() s.Unlock()
if !registered { if !registered {
log.Printf("Registering node: %s", node.Id) log.Logf("Registering node: %s", node.Id)
} }
// create registry options // create registry options
@ -327,7 +327,7 @@ func (s *rpcServer) Deregister() error {
Nodes: []*registry.Node{node}, Nodes: []*registry.Node{node},
} }
log.Printf("Deregistering node: %s", node.Id) log.Logf("Deregistering node: %s", node.Id)
if err := config.Registry.Deregister(service); err != nil { if err := config.Registry.Deregister(service); err != nil {
return err return err
} }
@ -343,7 +343,7 @@ func (s *rpcServer) Deregister() error {
for sb, subs := range s.subscribers { for sb, subs := range s.subscribers {
for _, sub := range subs { for _, sub := range subs {
log.Printf("Unsubscribing from topic: %s", sub.Topic()) log.Logf("Unsubscribing from topic: %s", sub.Topic())
sub.Unsubscribe() sub.Unsubscribe()
} }
s.subscribers[sb] = nil s.subscribers[sb] = nil
@ -362,7 +362,7 @@ func (s *rpcServer) Start() error {
return err return err
} }
log.Printf("Listening on %s", ts.Addr()) log.Logf("Listening on %s", ts.Addr())
s.Lock() s.Lock()
s.opts.Address = ts.Addr() s.opts.Address = ts.Addr()
s.Unlock() s.Unlock()

23
server/rpc_service.go
View File

@ -9,13 +9,13 @@ package server
import ( import (
"errors" "errors"
"io" "io"
"log"
"reflect" "reflect"
"strings" "strings"
"sync" "sync"
"unicode" "unicode"
"unicode/utf8" "unicode/utf8"
"github.com/micro/go-log"
"golang.org/x/net/context" "golang.org/x/net/context"
) )
@ -113,7 +113,7 @@ func prepareMethod(method reflect.Method) *methodType {
replyType = mtype.In(3) replyType = mtype.In(3)
contextType = mtype.In(1) contextType = mtype.In(1)
default: default:
log.Println("method", mname, "of", mtype, "has wrong number of ins:", mtype.NumIn()) log.Log("method", mname, "of", mtype, "has wrong number of ins:", mtype.NumIn())
return nil return nil
} }
@ -121,7 +121,7 @@ func prepareMethod(method reflect.Method) *methodType {
// check stream type // check stream type
streamType := reflect.TypeOf((*Streamer)(nil)).Elem() streamType := reflect.TypeOf((*Streamer)(nil)).Elem()
if !argType.Implements(streamType) { if !argType.Implements(streamType) {
log.Println(mname, "argument does not implement Streamer interface:", argType) log.Log(mname, "argument does not implement Streamer interface:", argType)
return nil return nil
} }
} else { } else {
@ -129,30 +129,30 @@ func prepareMethod(method reflect.Method) *methodType {
// First arg need not be a pointer. // First arg need not be a pointer.
if !isExportedOrBuiltinType(argType) { if !isExportedOrBuiltinType(argType) {
log.Println(mname, "argument type not exported:", argType) log.Log(mname, "argument type not exported:", argType)
return nil return nil
} }
if replyType.Kind() != reflect.Ptr { if replyType.Kind() != reflect.Ptr {
log.Println("method", mname, "reply type not a pointer:", replyType) log.Log("method", mname, "reply type not a pointer:", replyType)
return nil return nil
} }
// Reply type must be exported. // Reply type must be exported.
if !isExportedOrBuiltinType(replyType) { if !isExportedOrBuiltinType(replyType) {
log.Println("method", mname, "reply type not exported:", replyType) log.Log("method", mname, "reply type not exported:", replyType)
return nil return nil
} }
} }
// Method needs one out. // Method needs one out.
if mtype.NumOut() != 1 { if mtype.NumOut() != 1 {
log.Println("method", mname, "has wrong number of outs:", mtype.NumOut()) log.Log("method", mname, "has wrong number of outs:", mtype.NumOut())
return nil return nil
} }
// The return type of the method must be error. // The return type of the method must be error.
if returnType := mtype.Out(0); returnType != typeOfError { if returnType := mtype.Out(0); returnType != typeOfError {
log.Println("method", mname, "returns", returnType.String(), "not error") log.Log("method", mname, "returns", returnType.String(), "not error")
return nil return nil
} }
return &methodType{method: method, ArgType: argType, ReplyType: replyType, ContextType: contextType, stream: stream} return &methodType{method: method, ArgType: argType, ReplyType: replyType, ContextType: contextType, stream: stream}
@ -169,11 +169,12 @@ func (server *server) register(rcvr interface{}) error {
s.rcvr = reflect.ValueOf(rcvr) s.rcvr = reflect.ValueOf(rcvr)
sname := reflect.Indirect(s.rcvr).Type().Name() sname := reflect.Indirect(s.rcvr).Type().Name()
if sname == "" { if sname == "" {
log.Fatal("rpc: no service name for type", s.typ.String()) log.Log("rpc: no service name for type", s.typ.String())
return errors.New("rpc: no service name for type" + s.typ.String())
} }
if !isExported(sname) { if !isExported(sname) {
s := "rpc Register: type " + sname + " is not exported" s := "rpc Register: type " + sname + " is not exported"
log.Print(s) log.Log(s)
return errors.New(s) return errors.New(s)
} }
if _, present := server.serviceMap[sname]; present { if _, present := server.serviceMap[sname]; present {
@ -192,7 +193,7 @@ func (server *server) register(rcvr interface{}) error {
if len(s.method) == 0 { if len(s.method) == 0 {
s := "rpc Register: type " + sname + " has no exported methods of suitable type" s := "rpc Register: type " + sname + " has no exported methods of suitable type"
log.Print(s) log.Log(s)
return errors.New(s) return errors.New(s)
} }
server.serviceMap[s.name] = s server.serviceMap[s.name] = s

8
server/server.go
View File

@ -2,11 +2,11 @@
package server package server
import ( import (
"log"
"os" "os"
"os/signal" "os/signal"
"syscall" "syscall"
"github.com/micro/go-log"
"github.com/pborman/uuid" "github.com/pborman/uuid"
"golang.org/x/net/context" "golang.org/x/net/context"
) )
@ -140,7 +140,7 @@ func Run() error {
ch := make(chan os.Signal, 1) ch := make(chan os.Signal, 1)
signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL) signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
log.Printf("Received signal %s", <-ch) log.Logf("Received signal %s", <-ch)
if err := DefaultServer.Deregister(); err != nil { if err := DefaultServer.Deregister(); err != nil {
return err return err
@ -152,13 +152,13 @@ func Run() error {
// Start starts the default server // Start starts the default server
func Start() error { func Start() error {
config := DefaultServer.Options() config := DefaultServer.Options()
log.Printf("Starting server %s id %s", config.Name, config.Id) log.Logf("Starting server %s id %s", config.Name, config.Id)
return DefaultServer.Start() return DefaultServer.Start()
} }
// Stop stops the default server // Stop stops the default server
func Stop() error { func Stop() error {
log.Printf("Stopping server") log.Logf("Stopping server")
return DefaultServer.Stop() return DefaultServer.Stop()
} }

5
transport/http_transport.go
View File

@ -7,13 +7,13 @@ import (
"errors" "errors"
"io" "io"
"io/ioutil" "io/ioutil"
"log"
"net" "net"
"net/http" "net/http"
"net/url" "net/url"
"sync" "sync"
"time" "time"
"github.com/micro/go-log"
maddr "github.com/micro/misc/lib/addr" maddr "github.com/micro/misc/lib/addr"
mnet "github.com/micro/misc/lib/net" mnet "github.com/micro/misc/lib/net"
mls "github.com/micro/misc/lib/tls" mls "github.com/micro/misc/lib/tls"
@ -301,7 +301,7 @@ func (h *httpTransportListener) Accept(fn func(Socket)) error {
if max := 1 * time.Second; tempDelay > max { if max := 1 * time.Second; tempDelay > max {
tempDelay = max tempDelay = max
} }
log.Printf("http: Accept error: %v; retrying in %v\n", err, tempDelay) log.Logf("http: Accept error: %v; retrying in %v\n", err, tempDelay)
time.Sleep(tempDelay) time.Sleep(tempDelay)
continue continue
} }
@ -319,6 +319,7 @@ func (h *httpTransportListener) Accept(fn func(Socket)) error {
// TODO: think of a better error response strategy // TODO: think of a better error response strategy
defer func() { defer func() {
if r := recover(); r != nil { if r := recover(); r != nil {
log.Logf("http: recovered %v", r)
sock.Close() sock.Close()
} }
}() }()