Merge pull request #248 from micro/rework

Rework Interfaces

Author: Asim Aslam

client/backoff_test.go

@@ -10,8 +10,10 @@ import (
 func TestBackoff(t *testing.T) {
 	delta := time.Duration(0)
 
+	c := NewClient()
+
 	for i := 0; i < 5; i++ {
-		d, err := exponentialBackoff(context.TODO(), NewJsonRequest("test", "test", nil), i)
+		d, err := exponentialBackoff(context.TODO(), c.NewRequest("test", "test", nil), i)
 		if err != nil {
 			t.Fatal(err)
 		}

client/client.go

@@ -12,22 +12,18 @@ import (
 type Client interface {
 	Init(...Option) error
 	Options() Options
-	NewPublication(topic string, msg interface{}) Publication
+	NewMessage(topic string, msg interface{}) Message
 	NewRequest(service, method string, req interface{}, reqOpts ...RequestOption) Request
-	NewProtoRequest(service, method string, req interface{}, reqOpts ...RequestOption) Request
-	NewJsonRequest(service, method string, req interface{}, reqOpts ...RequestOption) Request
 	Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error
-	CallRemote(ctx context.Context, addr string, req Request, rsp interface{}, opts ...CallOption) error
-	Stream(ctx context.Context, req Request, opts ...CallOption) (Streamer, error)
-	StreamRemote(ctx context.Context, addr string, req Request, opts ...CallOption) (Streamer, error)
-	Publish(ctx context.Context, p Publication, opts ...PublishOption) error
+	Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error)
+	Publish(ctx context.Context, msg Message, opts ...PublishOption) error
 	String() string
 }
 
-// Publication is the interface for a message published asynchronously
-type Publication interface {
+// Message is the interface for publishing asynchronously
+type Message interface {
 	Topic() string
-	Message() interface{}
+	Payload() interface{}
 	ContentType() string
 }
 
@@ -41,8 +37,8 @@ type Request interface {
 	Stream() bool
 }
 
-// Streamer is the inteface for a bidirectional synchronous stream
-type Streamer interface {
+// Stream is the inteface for a bidirectional synchronous stream
+type Stream interface {
 	Context() context.Context
 	Request() Request
 	Send(interface{}) error
@@ -85,52 +81,32 @@ func Call(ctx context.Context, request Request, response interface{}, opts ...Ca
 	return DefaultClient.Call(ctx, request, response, opts...)
 }
 
-// Makes a synchronous call to the specified address using the default client
-func CallRemote(ctx context.Context, address string, request Request, response interface{}, opts ...CallOption) error {
-	return DefaultClient.CallRemote(ctx, address, request, response, opts...)
-}
-
-// Creates a streaming connection with a service and returns responses on the
-// channel passed in. It's up to the user to close the streamer.
-func Stream(ctx context.Context, request Request, opts ...CallOption) (Streamer, error) {
-	return DefaultClient.Stream(ctx, request, opts...)
-}
-
-// Creates a streaming connection to the address specified.
-func StreamRemote(ctx context.Context, address string, request Request, opts ...CallOption) (Streamer, error) {
-	return DefaultClient.StreamRemote(ctx, address, request, opts...)
-}
-
 // Publishes a publication using the default client. Using the underlying broker
 // set within the options.
-func Publish(ctx context.Context, p Publication) error {
-	return DefaultClient.Publish(ctx, p)
+func Publish(ctx context.Context, msg Message) error {
+	return DefaultClient.Publish(ctx, msg)
+}
+
+// Creates a new message using the default client
+func NewMessage(topic string, payload interface{}) Message {
+	return DefaultClient.NewMessage(topic, payload)
 }
 
 // Creates a new client with the options passed in
 func NewClient(opt ...Option) Client {
 	return newRpcClient(opt...)
 }
 
-// Creates a new publication using the default client
-func NewPublication(topic string, message interface{}) Publication {
-	return DefaultClient.NewPublication(topic, message)
-}
-
 // Creates a new request using the default client. Content Type will
 // be set to the default within options and use the appropriate codec
 func NewRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
 	return DefaultClient.NewRequest(service, method, request, reqOpts...)
 }
 
-// Creates a new protobuf request using the default client
-func NewProtoRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
-	return DefaultClient.NewProtoRequest(service, method, request, reqOpts...)
-}
-
-// Creates a new json request using the default client
-func NewJsonRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
-	return DefaultClient.NewJsonRequest(service, method, request, reqOpts...)
+// Creates a streaming connection with a service and returns responses on the
+// channel passed in. It's up to the user to close the streamer.
+func NewStream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
+	return DefaultClient.Stream(ctx, request, opts...)
 }
 
 func String() string {

client/mock/mock.go

@@ -49,22 +49,14 @@ func (m *MockClient) Options() client.Options {
 	return m.Opts
 }
 
-func (m *MockClient) NewPublication(topic string, msg interface{}) client.Publication {
-	return m.Client.NewPublication(topic, msg)
+func (m *MockClient) NewMessage(topic string, msg interface{}) client.Message {
+	return m.Client.NewMessage(topic, msg)
 }
 
 func (m *MockClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
 	return m.Client.NewRequest(service, method, req, reqOpts...)
 }
 
-func (m *MockClient) NewProtoRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
-	return m.Client.NewProtoRequest(service, method, req, reqOpts...)
-}
-
-func (m *MockClient) NewJsonRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
-	return m.Client.NewJsonRequest(service, method, req, reqOpts...)
-}
-
 func (m *MockClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
 	m.Lock()
 	defer m.Unlock()
@@ -97,55 +89,15 @@ func (m *MockClient) Call(ctx context.Context, req client.Request, rsp interface
 	return fmt.Errorf("rpc: can't find service %s", req.Method())
 }
 
-func (m *MockClient) CallRemote(ctx context.Context, addr string, req client.Request, rsp interface{}, opts ...client.CallOption) error {
-	m.Lock()
-	defer m.Unlock()
-
-	response, ok := m.Response[req.Service()]
-	if !ok {
-		return errors.NotFound("go.micro.client.mock", "service not found")
-	}
-
-	for _, r := range response {
-		if r.Method != req.Method() {
-			continue
-		}
-
-		if r.Error != nil {
-			return r.Error
-		}
-
-		v := reflect.ValueOf(rsp)
-
-		if t := reflect.TypeOf(rsp); t.Kind() == reflect.Ptr {
-			v = reflect.Indirect(v)
-		}
-
-		v.Set(reflect.ValueOf(r.Response))
-
-		return nil
-	}
-
-	return fmt.Errorf("rpc: can't find service %s", req.Method())
-}
-
-func (m *MockClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Streamer, error) {
-	m.Lock()
-	defer m.Unlock()
-
-	// TODO: mock stream
-	return nil, nil
-}
-
-func (m *MockClient) StreamRemote(ctx context.Context, addr string, req client.Request, opts ...client.CallOption) (client.Streamer, error) {
+func (m *MockClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
 	m.Lock()
 	defer m.Unlock()
 
 	// TODO: mock stream
 	return nil, nil
 }
 
-func (m *MockClient) Publish(ctx context.Context, p client.Publication, opts ...client.PublishOption) error {
+func (m *MockClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
 	return nil
 }
 

client/mock/mock_test.go

@@ -21,7 +21,7 @@ func TestClient(t *testing.T) {
 	c := NewClient(Response("go.mock", response))
 
 	for _, r := range response {
-		req := c.NewJsonRequest("go.mock", r.Method, map[string]interface{}{"foo": "bar"})
+		req := c.NewRequest("go.mock", r.Method, map[string]interface{}{"foo": "bar"})
 		var rsp interface{}
 
 		err := c.Call(context.TODO(), req, &rsp)

client/options.go

@@ -40,6 +40,8 @@ type Options struct {
 type CallOptions struct {
 	SelectOptions []selector.SelectOption
 
+	// Address of remote host
+	Address string
 	// Backoff func
 	Backoff BackoffFunc
 	// Check if retriable func
@@ -66,7 +68,8 @@ type PublishOptions struct {
 }
 
 type RequestOptions struct {
-	Stream bool
+	ContentType string
+	Stream      bool
 
 	// Other options for implementations of the interface
 	// can be stored in a context
@@ -226,6 +229,13 @@ func DialTimeout(d time.Duration) Option {
 
 // Call Options
 
+// WithAddress sets the remote address to use rather than using service discovery
+func WithAddress(a string) CallOption {
+	return func(o *CallOptions) {
+		o.Address = a
+	}
+}
+
 func WithSelectOption(so ...selector.SelectOption) CallOption {
 	return func(o *CallOptions) {
 		o.SelectOptions = append(o.SelectOptions, so...)
@@ -281,6 +291,12 @@ func WithDialTimeout(d time.Duration) CallOption {
 
 // Request Options
 
+func WithContentType(ct string) RequestOption {
+	return func(o *RequestOptions) {
+		o.ContentType = ct
+	}
+}
+
 func StreamingRequest() RequestOption {
 	return func(o *RequestOptions) {
 		o.Stream = true