grpc连接器和源信息（附加信息，header或者metate）

grpc的拦截器主要实现中间信息获取和统一错误处理，类似于java的filter
拦截器有两种，流式、非流式
非流式的函数签名：
func unaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error)
流式的函数签名
参考：https://github.com/grpc/grpc-go/tree/master/examples/features/interceptor
服务端代码示例：

package main
import (
    "context"
    "flag"
    "fmt"
    "io"
    "log"
    "net"
    "strings"
    "time"
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/credentials"
    ecpb "google.golang.org/grpc/examples/features/proto/echo"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
    "google.golang.org/grpc/testdata"
)
var (
    port = flag.Int("port", 50051, "the port to serve on")
    errMissingMetadata = status.Errorf(codes.InvalidArgument, "missing metadata")
    errInvalidToken    = status.Errorf(codes.Unauthenticated, "invalid token")
)
// logger is to mock a sophisticated logging system. To simplify the example, we just print out the content.
func logger(format string, a ...interface{}) {
    fmt.Printf("LOG:\t"+format+"\n", a...)
}
//实现接口----------
type server struct{}
func (s *server) UnaryEcho(ctx context.Context, in *ecpb.EchoRequest) (*ecpb.EchoResponse, error) {
    fmt.Printf("unary echoing message %q\n", in.Message)
    return &ecpb.EchoResponse{Message: in.Message}, nil
}
func (s *server) ServerStreamingEcho(in *ecpb.EchoRequest, stream ecpb.Echo_ServerStreamingEchoServer) error {
    return status.Error(codes.Unimplemented, "not implemented")
}
func (s *server) ClientStreamingEcho(stream ecpb.Echo_ClientStreamingEchoServer) error {
    return status.Error(codes.Unimplemented, "not implemented")
}
func (s *server) BidirectionalStreamingEcho(stream ecpb.Echo_BidirectionalStreamingEchoServer) error {
    for {
        in, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                return nil
            }
            fmt.Printf("server: error receiving from stream: %v\n", err)
            return err
        }
        fmt.Printf("bidi echoing message %q\n", in.Message)
        stream.Send(&ecpb.EchoResponse{Message: in.Message})
    }
}
//实现接口------------
// valid validates the authorization.
func valid(authorization []string) bool {
    if len(authorization) < 1 {
        return false
    }
    token := strings.TrimPrefix(authorization[0], "Bearer ")
    // Perform the token validation here. For the sake of this example, the code
    // here forgoes any of the usual OAuth2 token validation and instead checks
    // for a token matching an arbitrary string.
    return token == "some-secret-token"
}
//非流式的拦截器
func unaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    // authentication (token verification)
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return nil, errMissingMetadata
    }
    if !valid(md["authorization"]) {
        return nil, errInvalidToken
    }
    m, err := handler(ctx, req)
    if err != nil {
        logger("RPC failed with error %v", err)
    }
    return m, err
}
// wrappedStream wraps around the embedded grpc.ServerStream, and intercepts the RecvMsg and
// SendMsg method call.
type wrappedStream struct {
    grpc.ServerStream
}
func (w *wrappedStream) RecvMsg(m interface{}) error {
    logger("Receive a message (Type: %T) at %s", m, time.Now().Format(time.RFC3339))
    return w.ServerStream.RecvMsg(m)
}
func (w *wrappedStream) SendMsg(m interface{}) error {
    logger("Send a message (Type: %T) at %v", m, time.Now().Format(time.RFC3339))
    return w.ServerStream.SendMsg(m)
}
func newWrappedStream(s grpc.ServerStream) grpc.ServerStream {
    return &wrappedStream{s}
}
func streamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    // authentication (token verification)
    md, ok := metadata.FromIncomingContext(ss.Context())
    if !ok {
        return errMissingMetadata
    }
    if !valid(md["authorization"]) {
        return errInvalidToken
    }
    err := handler(srv, newWrappedStream(ss))
    if err != nil {
        logger("RPC failed with error %v", err)
    }
    return err
}
func main() {
    flag.Parse()
    lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    // Create tls based credential.
    creds, err := credentials.NewServerTLSFromFile(testdata.Path("server1.pem"), testdata.Path("server1.key"))
    if err != nil {
        log.Fatalf("failed to create credentials: %v", err)
    }
    s := grpc.NewServer(grpc.Creds(creds), grpc.UnaryInterceptor(unaryInterceptor), grpc.StreamInterceptor(streamInterceptor))
    // Register EchoServer on the server.
    ecpb.RegisterEchoServer(s, &server{})
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

package main
import (
    "context"
    "flag"
    "fmt"
    "io"
    "log"
    "time"
    "golang.org/x/oauth2"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/credentials/oauth"
    ecpb "google.golang.org/grpc/examples/features/proto/echo"
    "google.golang.org/grpc/testdata"
)
var addr = flag.String("addr", "localhost:50051", "the address to connect to")
const fallbackToken = "some-secret-token"
// logger is to mock a sophisticated logging system. To simplify the example, we just print out the content.
func logger(format string, a ...interface{}) {
    fmt.Printf("LOG:\t"+format+"\n", a...)
}
// unaryInterceptor is an example unary interceptor.
func unaryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
    var credsConfigured bool
    for _, o := range opts {
        _, ok := o.(grpc.PerRPCCredsCallOption)
        if ok {
            credsConfigured = true
            break
        }
    }
    if !credsConfigured {
        opts = append(opts, grpc.PerRPCCredentials(oauth.NewOauthAccess(&oauth2.Token{
            AccessToken: fallbackToken,
        })))
    }
    start := time.Now()
    err := invoker(ctx, method, req, reply, cc, opts...)
    end := time.Now()
    logger("RPC: %s, start time: %s, end time: %s, err: %v", method, start.Format("Basic"), end.Format(time.RFC3339), err)
    return err
}
// wrappedStream  wraps around the embedded grpc.ClientStream, and intercepts the RecvMsg and
// SendMsg method call.
type wrappedStream struct {
    grpc.ClientStream
}
func (w *wrappedStream) RecvMsg(m interface{}) error {
    logger("Receive a message (Type: %T) at %v", m, time.Now().Format(time.RFC3339))
    return w.ClientStream.RecvMsg(m)
}
func (w *wrappedStream) SendMsg(m interface{}) error {
    logger("Send a message (Type: %T) at %v", m, time.Now().Format(time.RFC3339))
    return w.ClientStream.SendMsg(m)
}
func newWrappedStream(s grpc.ClientStream) grpc.ClientStream {
    return &wrappedStream{s}
}
// streamInterceptor is an example stream interceptor.
func streamInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
    var credsConfigured bool
    for _, o := range opts {
        _, ok := o.(*grpc.PerRPCCredsCallOption)
        if ok {
            credsConfigured = true
        }
    }
    if !credsConfigured {
        opts = append(opts, grpc.PerRPCCredentials(oauth.NewOauthAccess(&oauth2.Token{
            AccessToken: fallbackToken,
        })))
    }
    s, err := streamer(ctx, desc, cc, method, opts...)
    if err != nil {
        return nil, err
    }
    return newWrappedStream(s), nil
}
func callUnaryEcho(client ecpb.EchoClient, message string) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    resp, err := client.UnaryEcho(ctx, &ecpb.EchoRequest{Message: message})
    if err != nil {
        log.Fatalf("client.UnaryEcho(_) = _, %v: ", err)
    }
    fmt.Println("UnaryEcho: ", resp.Message)
}
func callBidiStreamingEcho(client ecpb.EchoClient) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    c, err := client.BidirectionalStreamingEcho(ctx)
    if err != nil {
        return
    }
    for i := 0; i < 5; i++ {
        if err := c.Send(&ecpb.EchoRequest{Message: fmt.Sprintf("Request %d", i+1)}); err != nil {
            log.Fatalf("failed to send request due to error: %v", err)
        }
    }
    c.CloseSend()
    for {
        resp, err := c.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            log.Fatalf("failed to receive response due to error: %v", err)
        }
        fmt.Println("BidiStreaming Echo: ", resp.Message)
    }
}
func main() {
    flag.Parse()
    // Create tls based credential.
    creds, err := credentials.NewClientTLSFromFile(testdata.Path("ca.pem"), "x.test.youtube.com")
    if err != nil {
        log.Fatalf("failed to load credentials: %v", err)
    }
    // Set up a connection to the server.
    conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(creds), grpc.WithUnaryInterceptor(unaryInterceptor), grpc.WithStreamInterceptor(streamInterceptor))
    if err != nil {
        log.Fatalf("did not connect: %v", err)
    }
    defer conn.Close()
    // Make a echo client and send RPCs.
    rgc := ecpb.NewEchoClient(conn)
    callUnaryEcho(rgc, "hello world")
    callBidiStreamingEcho(rgc)
}