remoting

Documentation

Overview

* Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License.

* Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License.

* Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License.

* Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License.

Index

• Constants
• func AddPendingResponse(pr *PendingResponse)
• func RegistryCodec(protocol string, codecTmp Codec)
• func SequenceID() int64
• type AsyncCallbackResponse
• type Client
• type Codec
  • func GetCodec(protocol string) Codec
• type DataListener
• type DecodeResult
• type Event
  • func (e Event) String() string
• type EventType
  • func (t EventType) String() string
• type ExchangeClient
  • func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Duration, lazyInit bool) *ExchangeClient
  • func (client *ExchangeClient) AsyncRequest(invocation *protocol.Invocation, url *common.URL, timeout time.Duration, ...) error
  • func (client *ExchangeClient) Close()
  • func (client *ExchangeClient) DecreaseActiveNumber() uint32
  • func (client *ExchangeClient) GetActiveNumber() uint32
  • func (client *ExchangeClient) IncreaseActiveNumber() uint32
  • func (client *ExchangeClient) IsAvailable() bool
  • func (client *ExchangeClient) Request(invocation *protocol.Invocation, url *common.URL, timeout time.Duration, ...) error
  • func (client *ExchangeClient) Send(invocation *protocol.Invocation, url *common.URL, timeout time.Duration) error
• type ExchangeServer
  • func NewExchangeServer(url *common.URL, server Server) *ExchangeServer
  • func (server *ExchangeServer) Start()
  • func (server *ExchangeServer) Stop()
• type Options
• type PendingResponse
  • func GetPendingResponse(seq SequenceType) *PendingResponse
  • func NewPendingResponse(id int64) *PendingResponse
  • func (r PendingResponse) GetCallResponse() common.CallbackResponse
  • func (r *PendingResponse) SetResponse(response *Response)
• type Request
  • func NewRequest(version string) *Request
• type Response
  • func NewResponse(id int64, version string) *Response
  • func (response *Response) Handle()
  • func (response *Response) IsHeartbeat() bool
• type SequenceType
• type Server

Constants

const (
	// EventTypeAdd means add event
	EventTypeAdd = iota
	// EventTypeDel means del event
	EventTypeDel
	// EventTypeUpdate means update event
	EventTypeUpdate
)

Functions

func AddPendingResponse

func AddPendingResponse(pr *PendingResponse)

store response into map

func RegistryCodec

func RegistryCodec(protocol string, codecTmp Codec)

func SequenceID

func SequenceID() int64

Types

type AsyncCallbackResponse

type AsyncCallbackResponse struct {
	common.CallbackResponse
	Opts      Options
	Cause     error
	Start     time.Time // invoke(call) start time == write start time
	ReadStart time.Time // read start time, write duration = ReadStart - Start
	Reply     interface{}
}

AsyncCallbackResponse async response for dubbo

type Client

type Client interface {
	SetExchangeClient(client *ExchangeClient)
	// connect url
	Connect(url *common.URL) error
	// close
	Close()
	// send request to server.
	Request(request *Request, timeout time.Duration, response *PendingResponse) error
	// check if the client is still available
	IsAvailable() bool
}

It is interface of client for network communication. If you use getty as network communication, you should define GettyClient that implements this interface.

type Codec

type Codec interface {
	EncodeRequest(request *Request) (*bytes.Buffer, error)
	EncodeResponse(response *Response) (*bytes.Buffer, error)
	Decode(data []byte) (DecodeResult, int, error)
}

codec for exchangeClient

func GetCodec

func GetCodec(protocol string) Codec

type DataListener

type DataListener interface {
	DataChange(eventType Event) bool // bool is return for interface implement is interesting
}

DataListener defines common data listener interface

type DecodeResult

type DecodeResult struct {
	IsRequest bool
	Result    interface{}
}

type Event

type Event struct {
	Path    string
	Action  EventType
	Content string
}

Event defines common elements for service event

func (Event) String

func (e Event) String() string

nolint

type EventType

type EventType int

EventType means SourceObjectEventType

func (EventType) String

func (t EventType) String() string

nolint

type ExchangeClient

type ExchangeClient struct {
	// connect server timeout
	ConnectTimeout time.Duration
	// contains filtered or unexported fields
}

This is abstraction level. it is like facade.

func NewExchangeClient

func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Duration, lazyInit bool) *ExchangeClient

create ExchangeClient

func (*ExchangeClient) AsyncRequest

func (client *ExchangeClient) AsyncRequest(invocation *protocol.Invocation, url *common.URL, timeout time.Duration,
	callback common.AsyncCallback, result *protocol.RPCResult) error

async two way request

func (*ExchangeClient) Close

func (client *ExchangeClient) Close()

close client

func (*ExchangeClient) DecreaseActiveNumber

func (client *ExchangeClient) DecreaseActiveNumber() uint32

decrease number of service using client

func (*ExchangeClient) GetActiveNumber

func (client *ExchangeClient) GetActiveNumber() uint32

get number of service using client

func (*ExchangeClient) IncreaseActiveNumber

func (client *ExchangeClient) IncreaseActiveNumber() uint32

increase number of service using client

func (*ExchangeClient) IsAvailable

func (client *ExchangeClient) IsAvailable() bool

IsAvailable to check if the underlying network client is available yet.

func (*ExchangeClient) Request

func (client *ExchangeClient) Request(invocation *protocol.Invocation, url *common.URL, timeout time.Duration,
	result *protocol.RPCResult) error

two way request

func (*ExchangeClient) Send

func (client *ExchangeClient) Send(invocation *protocol.Invocation, url *common.URL, timeout time.Duration) error

oneway request

type ExchangeServer

type ExchangeServer struct {
	Server Server
	URL    *common.URL
}

This is abstraction level. it is like facade.

func NewExchangeServer

func NewExchangeServer(url *common.URL, server Server) *ExchangeServer

Create ExchangeServer

func (*ExchangeServer) Start

func (server *ExchangeServer) Start()

start server

func (*ExchangeServer) Stop

func (server *ExchangeServer) Stop()

stop server

type Options

type Options struct {
	// connect timeout
	ConnectTimeout time.Duration
}

type PendingResponse

type PendingResponse struct {
	Err error

	ReadStart time.Time
	Callback  common.AsyncCallback

	Reply interface{}
	Done  chan struct{}
	// contains filtered or unexported fields
}

the client sends request to server, there is one pendingResponse at client side to wait the response from server

func GetPendingResponse

func GetPendingResponse(seq SequenceType) *PendingResponse

get response

func NewPendingResponse

func NewPendingResponse(id int64) *PendingResponse

NewPendingResponse aims to create PendingResponse. ID is always from ID of Request

func (PendingResponse) GetCallResponse

func (r PendingResponse) GetCallResponse() common.CallbackResponse

GetCallResponse is used for callback of async. It is will return AsyncCallbackResponse.

func (*PendingResponse) SetResponse

func (r *PendingResponse) SetResponse(response *Response)

type Request

type Request struct {
	ID int64
	// protocol version
	Version string
	// serial ID (ignore)
	SerialID byte
	// Data
	Data   interface{}
	TwoWay bool
	Event  bool
}

this is request for transport layer

func NewRequest

func NewRequest(version string) *Request

NewRequest aims to create Request. The ID is auto increase.

type Response

type Response struct {
	ID       int64
	Version  string
	SerialID byte
	Status   uint8
	Event    bool
	Error    error
	Result   interface{}
}

this is response for transport layer

func NewResponse

func NewResponse(id int64, version string) *Response

NewResponse create to a new Response.

func (*Response) Handle

func (response *Response) Handle()

func (*Response) IsHeartbeat

func (response *Response) IsHeartbeat() bool

the response is heartbeat

type SequenceType

type SequenceType int64

type Server

type Server interface {
	// invoke once for connection
	Start()
	// it is for destroy
	Stop()
}

It is interface of server for network communication. If you use getty as network communication, you should define GettyServer that implements this interface.