[ISSIUE #75] support shutdown mehtod #97
Conversation
2: add InvokeAsync method unit tests 3: change some fields visibility
internal/kernel/client.go
Outdated
| } | ||
| }() | ||
| }) | ||
| } | ||
|
|
||
| func (c *RMQClient) Shutdown() { | ||
| // TODO |
There was a problem hiding this comment.
it seems that if I shutdown the RMQClient, I remain send message by RMQClient.
There was a problem hiding this comment.
Yes, but it cancel four goroutines in the Start methods.
There was a problem hiding this comment.
Yes, but it cancels four goroutines in the Start method.
There was a problem hiding this comment.
IMO, could we add a field like
stateinstead ofcancel, which is updated toshutdownwhenShutdown()called, and return a error likeerrors.New("client closed")whenSend*()called
How about this?
internal/kernel/client.go
Outdated
| @@ -95,6 +95,7 @@ type RMQClient struct { | |||
|
|
|||
| remoteClient *remote.RemotingClient | |||
| hbMutex sync.Mutex | |||
| cancel context.CancelFunc | |||
There was a problem hiding this comment.
IMO, could we add a field like state instead of cancel, which is updated to shutdown when Shutdown() called, and return a error like errors.New("client closed") when Send*() called
There was a problem hiding this comment.
OK，I will adjust the implementation and make it sensible.
| @@ -367,3 +368,64 @@ func TestInvokeOneWay(t *testing.T) { | |||
| } | |||
| wg.Done() | |||
| } | |||
|
|
|||
|
|
|||
| func TestInvokeAsync2(t *testing.T) { | |||
internal/kernel/client.go
Outdated
| @@ -95,6 +95,7 @@ type RMQClient struct { | |||
|
|
|||
| remoteClient *remote.RemotingClient | |||
| hbMutex sync.Mutex | |||
| state bool | |||
There was a problem hiding this comment.
use close make more readable
internal/kernel/client.go
Outdated
| @@ -125,44 +129,64 @@ func (c *RMQClient) Start() { | |||
| // delay | |||
| time.Sleep(50 * time.Millisecond) | |||
| for { | |||
There was a problem hiding this comment.
below is better?
for !c.close {
c.UpdateTopicRouteInfo()
time.Sleep(_PullNameServerInterval)
}
internal/kernel/client.go
Outdated
| time.Sleep(_PullNameServerInterval) | ||
| } else { | ||
| return | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| // TODO cleanOfflineBroker & sendHeartbeatToAllBrokerWithLock | ||
| go func() { | ||
| for { |
There was a problem hiding this comment.
below is better?
for !c.close { c.UpdateTopicRouteInfo() time.Sleep(_PullNameServerInterval) }
internal/kernel/client.go
Outdated
| } else { | ||
| return | ||
| } | ||
|
|
||
| } | ||
| }() | ||
|
|
||
| // schedule persist offset | ||
| go func() { | ||
| //time.Sleep(10 * time.Second) | ||
| for { |
There was a problem hiding this comment.
below is better?
for !c.close { c.UpdateTopicRouteInfo() time.Sleep(_PullNameServerInterval) }
internal/kernel/client.go
Outdated
| } | ||
| }() | ||
|
|
||
| go func() { | ||
| for { | ||
| c.RebalanceImmediately() |
There was a problem hiding this comment.
below is better?
for !c.close { c.UpdateTopicRouteInfo() time.Sleep(_PullNameServerInterval) }
internal/kernel/client.go
Outdated
| return c.remoteClient.InvokeSync(addr, request, timeoutMillis) | ||
| if c.state { | ||
| return c.remoteClient.InvokeSync(addr, request, timeoutMillis) | ||
| } else { |
There was a problem hiding this comment.
else is not needed.
if c.close {
return nil, ErrServiceState
}
return return c.remoteClient.InvokeSync(addr, request, timeoutMillis)
internal/kernel/client.go
Outdated
| return c.remoteClient.InvokeAsync(addr, request, timeoutMillis, func(future *remote.ResponseFuture) { | ||
| f(future.ResponseCommand, future.Err) | ||
| }) | ||
| } else { |
There was a problem hiding this comment.
elseis not needed.if c.close { return nil, ErrServiceState } return return c.remoteClient.InvokeSync(addr, request, timeoutMillis)
internal/kernel/client.go
Outdated
| if c.state { | ||
| return c.remoteClient.InvokeOneWay(addr, request, timeoutMillis) | ||
|
|
||
| } else { |
There was a problem hiding this comment.
elseis not needed.if c.close { return nil, ErrServiceState } return return c.remoteClient.InvokeSync(addr, request, timeoutMillis)
gaufung
changed the title
|
LGTM, @ShannonDing @zongtanghu @jonnxu please merge the PR |
What is the purpose of the change
fix remote timeout uint and add
shutdownmethodBrief changelog
RemotingClient'sshutdownmethodVerifying this change
XXXX
Follow this checklist to help us incorporate your contribution quickly and easily. Notice,
it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.[ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.