From 3e08d413cdb67d3632f0b1462a1b306ad446e6b7 Mon Sep 17 00:00:00 2001 From: AlexStocks Date: Sun, 3 Jan 2021 03:36:55 +0800 Subject: [PATCH] fix remoting linter error --- cluster/router/chain/chain_test.go | 9 +++- config_center/zookeeper/impl_test.go | 20 +++++++-- metadata/report/delegate/delegate_report.go | 19 +++++++-- metadata/report/etcd/report_test.go | 5 ++- .../service/inmemory/service_proxy_test.go | 19 ++++++--- metadata/service/inmemory/service_test.go | 41 +++++++++++++------ metadata/service/remote/service_proxy_test.go | 34 +++++++++------ protocol/dubbo/hessian2/hessian_request.go | 17 +++++--- registry/file/service_discovery.go | 4 +- registry/file/service_discovery_test.go | 12 ++++-- remoting/getty/getty_client_test.go | 7 +++- remoting/getty/pool.go | 20 ++++++--- remoting/zookeeper/client_test.go | 25 ++++++++--- remoting/zookeeper/facade_test.go | 6 ++- remoting/zookeeper/listener_test.go | 6 ++- 15 files changed, 179 insertions(+), 65 deletions(-) diff --git a/cluster/router/chain/chain_test.go b/cluster/router/chain/chain_test.go index b21990b08c..0c3b2187e6 100644 --- a/cluster/router/chain/chain_test.go +++ b/cluster/router/chain/chain_test.go @@ -133,6 +133,7 @@ conditions: zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl) + assert.NoError(t, err) config.GetEnvInstance().SetDynamicConfiguration(configuration) chain, err := NewRouterChain(getConditionRouteUrl(applicationKey)) @@ -154,11 +155,15 @@ conditions: func TestRouterChainRoute(t *testing.T) { ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second) - defer ts.Stop() - defer z.Close() + defer func() { + err = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl) + assert.NoError(t, err) config.GetEnvInstance().SetDynamicConfiguration(configuration) chain, err := NewRouterChain(getConditionRouteUrl(applicationKey)) diff --git a/config_center/zookeeper/impl_test.go b/config_center/zookeeper/impl_test.go index 808bf8fd57..a518bb6ec0 100644 --- a/config_center/zookeeper/impl_test.go +++ b/config_center/zookeeper/impl_test.go @@ -99,7 +99,10 @@ func initZkData(group string, t *testing.T) (*zk.TestCluster, *zookeeperDynamicC func TestGetConfig(t *testing.T) { ts, reg := initZkData("dubbo", t) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.NoError(t, err) + }() configs, err := reg.GetProperties(dubboPropertyFileName, config_center.WithGroup("dubbo")) assert.NoError(t, err) m, err := reg.Parser().Parse(configs) @@ -118,7 +121,10 @@ func TestGetConfig(t *testing.T) { func TestAddListener(t *testing.T) { ts, reg := initZkData("", t) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.NoError(t, err) + }() listener := &mockDataListener{} reg.AddListener(dubboPropertyFileName, listener) listener.wg.Add(1) @@ -151,7 +157,10 @@ func TestAddListener(t *testing.T) { func TestRemoveListener(t *testing.T) { ts, reg := initZkData("", t) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.NoError(t, err) + }() listener := &mockDataListener{} reg.AddListener(dubboPropertyFileName, listener) listener.wg.Add(1) @@ -189,7 +198,10 @@ func TestZookeeperDynamicConfigurationPublishConfig(t *testing.T) { customGroup := "Custom Group" key := "myKey" ts, zk := initZkData(config_center.DEFAULT_GROUP, t) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.NoError(t, err) + }() err := zk.PublishConfig(key, customGroup, value) assert.Nil(t, err) result, err := zk.GetInternalProperty("myKey", config_center.WithGroup(customGroup)) diff --git a/metadata/report/delegate/delegate_report.go b/metadata/report/delegate/delegate_report.go index 836a8f9ef4..f364087fdc 100644 --- a/metadata/report/delegate/delegate_report.go +++ b/metadata/report/delegate/delegate_report.go @@ -220,7 +220,11 @@ func (mr *MetadataReport) SaveServiceMetadata(identifier *identifier.ServiceMeta if mr.syncReport { return report.SaveServiceMetadata(identifier, url) } - go report.SaveServiceMetadata(identifier, url) + go func() { + if err := report.SaveServiceMetadata(identifier, url); err != nil { + logger.Warnf("report.SaveServiceMetadata(identifier:%v, url:%v) = error:%v", identifier, url, err) + } + }() return nil } @@ -230,7 +234,11 @@ func (mr *MetadataReport) RemoveServiceMetadata(identifier *identifier.ServiceMe if mr.syncReport { return report.RemoveServiceMetadata(identifier) } - go report.RemoveServiceMetadata(identifier) + go func() { + if err := report.RemoveServiceMetadata(identifier); err != nil { + logger.Warnf("report.RemoveServiceMetadata(identifier:%v) = error:%v", identifier, err) + } + }() return nil } @@ -255,7 +263,12 @@ func (mr *MetadataReport) SaveSubscribedData(identifier *identifier.SubscriberMe if mr.syncReport { return report.SaveSubscribedData(identifier, string(bytes)) } - go report.SaveSubscribedData(identifier, string(bytes)) + go func() { + if err := report.SaveSubscribedData(identifier, string(bytes)); err != nil { + logger.Warnf("report.SaveSubscribedData(identifier:%v, string(bytes):%v) = error: %v", + identifier, string(bytes), err) + } + }() return nil } diff --git a/metadata/report/etcd/report_test.go b/metadata/report/etcd/report_test.go index 28c0486982..59d0975ca6 100644 --- a/metadata/report/etcd/report_test.go +++ b/metadata/report/etcd/report_test.go @@ -82,8 +82,9 @@ func TestEtcdMetadataReport_CRUD(t *testing.T) { assert.Nil(t, err) serviceMi := newServiceMetadataIdentifier() - serviceUrl, _ := common.NewURL("registry://localhost:8848", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) - metadataReport.SaveServiceMetadata(serviceMi, serviceUrl) + serviceUrl, err := common.NewURL("registry://localhost:8848", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) + assert.Nil(t, err) + err = metadataReport.SaveServiceMetadata(serviceMi, serviceUrl) assert.Nil(t, err) subMi := newSubscribeMetadataIdentifier() diff --git a/metadata/service/inmemory/service_proxy_test.go b/metadata/service/inmemory/service_proxy_test.go index 01d9206f43..9278fd9d6c 100644 --- a/metadata/service/inmemory/service_proxy_test.go +++ b/metadata/service/inmemory/service_proxy_test.go @@ -62,12 +62,19 @@ func TestNewMetadataService(t *testing.T) { assert.Nil(t, err) _, err = pxy.GetServiceDefinitionByServiceKey("any") assert.Nil(t, err) - pxy.ExportURL(&common.URL{}) - pxy.SubscribeURL(&common.URL{}) - pxy.MethodMapper() - pxy.UnexportURL(&common.URL{}) - pxy.RefreshMetadata(constant.ANY_VALUE, constant.ANY_VALUE) - + ok, err := pxy.ExportURL(&common.URL{}) + assert.False(t, ok) + assert.NoError(t, err) + ok, err = pxy.SubscribeURL(&common.URL{}) + assert.False(t, ok) + assert.NoError(t, err) + m := pxy.MethodMapper() + assert.True(t, len(m) == 0) + err = pxy.UnexportURL(&common.URL{}) + assert.NoError(t, err) + ok, err = pxy.RefreshMetadata(constant.ANY_VALUE, constant.ANY_VALUE) + assert.False(t, ok) + assert.NoError(t, err) } func createPxy() service.MetadataService { diff --git a/metadata/service/inmemory/service_test.go b/metadata/service/inmemory/service_test.go index 256412c291..e50cd6208a 100644 --- a/metadata/service/inmemory/service_test.go +++ b/metadata/service/inmemory/service_test.go @@ -46,7 +46,9 @@ func TestMetadataService(t *testing.T) { "owner=ZX&pid=1447&revision=0.0.1&side=provider&timeout=3000×tamp=1556509797245&group=%v&version=%v&bean.name=%v", protocol, serviceName, group, version, beanName)) assert.NoError(t, err) - mts.ExportURL(u2) + ok, err := mts.ExportURL(u2) + assert.True(t, ok) + assert.NoError(t, err) u3, err := common.NewURL(fmt.Sprintf( "%v://127.0.0.1:20000/com.ikurento.user.UserProvider3?anyhost=true&"+ @@ -55,7 +57,9 @@ func TestMetadataService(t *testing.T) { "owner=ZX&pid=1447&revision=0.0.1&side=provider&timeout=3000×tamp=1556509797245&group=%v&version=%v&bean.name=%v", protocol, serviceName, group, version, beanName)) assert.NoError(t, err) - mts.ExportURL(u3) + ok, err = mts.ExportURL(u3) + assert.True(t, ok) + assert.NoError(t, err) u, err := common.NewURL(fmt.Sprintf( "%v://127.0.0.1:20000/com.ikurento.user.UserProvider1?anyhost=true&"+ @@ -64,32 +68,45 @@ func TestMetadataService(t *testing.T) { "owner=ZX&pid=1447&revision=0.0.1&side=provider&timeout=3000×tamp=1556509797245&group=%v&version=%v&bean.name=%v", protocol, serviceName, group, version, beanName)) assert.NoError(t, err) - mts.ExportURL(u) + ok, err = mts.ExportURL(u) + assert.True(t, ok) + assert.NoError(t, err) list, _ := mts.GetExportedURLs(serviceName, group, version, protocol) assert.Equal(t, 3, len(list)) - mts.SubscribeURL(u) + ok, err = mts.SubscribeURL(u) + assert.True(t, ok) + assert.NoError(t, err) - mts.SubscribeURL(u) - list2, _ := mts.GetSubscribedURLs() + ok, err = mts.SubscribeURL(u) + assert.False(t, ok) + assert.NoError(t, err) + list2, err := mts.GetSubscribedURLs() assert.Equal(t, 1, len(list2)) + assert.NoError(t, err) - mts.UnexportURL(u) + err = mts.UnexportURL(u) + assert.NoError(t, err) list3, _ := mts.GetExportedURLs(serviceName, group, version, protocol) assert.Equal(t, 2, len(list3)) - mts.UnsubscribeURL(u) + err = mts.UnsubscribeURL(u) + assert.NoError(t, err) list4, _ := mts.GetSubscribedURLs() assert.Equal(t, 0, len(list4)) userProvider := &definition.UserProvider{} - common.ServiceMap.Register(serviceName, protocol, group, version, userProvider) - mts.PublishServiceDefinition(u) + _, err = common.ServiceMap.Register(serviceName, protocol, group, version, userProvider) + assert.NoError(t, err) + err = mts.PublishServiceDefinition(u) + assert.NoError(t, err) expected := "{\"CanonicalName\":\"com.ikurento.user.UserProvider\",\"CodeSource\":\"\"," + "\"Methods\":[{\"Name\":\"GetUser\",\"ParameterTypes\":[\"slice\"],\"ReturnType\":\"ptr\"," + "\"Parameters\":null}],\"Types\":null}" - def1, _ := mts.GetServiceDefinition(serviceName, group, version) + def1, err := mts.GetServiceDefinition(serviceName, group, version) assert.Equal(t, expected, def1) + assert.NoError(t, err) serviceKey := definition.ServiceDescriperBuild(serviceName, group, version) - def2, _ := mts.GetServiceDefinitionByServiceKey(serviceKey) + def2, err := mts.GetServiceDefinitionByServiceKey(serviceKey) assert.Equal(t, expected, def2) + assert.NoError(t, err) } diff --git a/metadata/service/remote/service_proxy_test.go b/metadata/service/remote/service_proxy_test.go index 8bccbb8bbb..1899d02205 100644 --- a/metadata/service/remote/service_proxy_test.go +++ b/metadata/service/remote/service_proxy_test.go @@ -54,18 +54,28 @@ func TestMetadataServiceProxy_GetServiceDefinition(t *testing.T) { // in fact, we don't use them func TestMetadataServiceProxy(t *testing.T) { pxy := createProxy() - pxy.ServiceName() - pxy.PublishServiceDefinition(&common.URL{}) - pxy.Version() - pxy.GetSubscribedURLs() - pxy.UnsubscribeURL(&common.URL{}) - pxy.GetServiceDefinitionByServiceKey("any") - pxy.ExportURL(&common.URL{}) - pxy.SubscribeURL(&common.URL{}) - pxy.MethodMapper() - pxy.UnexportURL(&common.URL{}) - pxy.Reference() - pxy.RefreshMetadata(constant.ANY_VALUE, constant.ANY_VALUE) + _, err := pxy.ServiceName() + assert.NoError(t, err) + err = pxy.PublishServiceDefinition(&common.URL{}) + assert.NoError(t, err) + _, err = pxy.Version() + assert.NoError(t, err) + _, err = pxy.GetSubscribedURLs() + assert.NoError(t, err) + err = pxy.UnsubscribeURL(&common.URL{}) + assert.NoError(t, err) + _, err = pxy.GetServiceDefinitionByServiceKey("any") + assert.NoError(t, err) + _, err = pxy.ExportURL(&common.URL{}) + assert.NoError(t, err) + _, err = pxy.SubscribeURL(&common.URL{}) + assert.NoError(t, err) + _ = pxy.MethodMapper() + err = pxy.UnexportURL(&common.URL{}) + assert.NoError(t, err) + _ = pxy.Reference() + _, err = pxy.RefreshMetadata(constant.ANY_VALUE, constant.ANY_VALUE) + assert.NoError(t, err) } func createProxy() service.MetadataService { diff --git a/protocol/dubbo/hessian2/hessian_request.go b/protocol/dubbo/hessian2/hessian_request.go index 2a1d5f736d..af603014c8 100644 --- a/protocol/dubbo/hessian2/hessian_request.go +++ b/protocol/dubbo/hessian2/hessian_request.go @@ -220,29 +220,34 @@ func packRequest(service Service, header DubboHeader, req interface{}) ([]byte, // body ////////////////////////////////////////// if hb { - encoder.Encode(nil) + if err := encoder.Encode(nil); err != nil { + logger.Warnf("Encode(nil) = error: %v", err) + } goto END } // dubbo version + path + version + method if err := encoder.Encode(DEFAULT_DUBBO_PROTOCOL_VERSION); err != nil { - logger.Error("Encode(DEFAULT_DUBBO_PROTOCOL_VERSION) = error: %v", err) + logger.Warnf("Encode(DEFAULT_DUBBO_PROTOCOL_VERSION) = error: %v", err) } if err := encoder.Encode(service.Path); err != nil { - logger.Error("Encode(service.Path) = error: %v", err) + logger.Warnf("Encode(service.Path) = error: %v", err) } if err := encoder.Encode(service.Version); err != nil { - logger.Error("Encode(service.Version) = error: %v", err) + logger.Warnf("Encode(service.Version) = error: %v", err) } if err := encoder.Encode(service.Method); err != nil { - logger.Error("Encode(service.Method) = error: %v", err) + logger.Warnf("Encode(service.Method) = error: %v", err) } // args = args type list + args value list if types, err = getArgsTypeList(args); err != nil { return nil, perrors.Wrapf(err, " PackRequest(args:%+v)", args) } - encoder.Encode(types) + if err := encoder.Encode(types); err != nil { + logger.Warnf("Encode(types:%v) = error: %v", types, err) + } + for _, v := range args { encoder.Encode(v) } diff --git a/registry/file/service_discovery.go b/registry/file/service_discovery.go index 254c12688f..d19d1f310c 100644 --- a/registry/file/service_discovery.go +++ b/registry/file/service_discovery.go @@ -82,7 +82,9 @@ func newFileSystemServiceDiscovery(name string) (registry.ServiceDiscovery, erro } extension.AddCustomShutdownCallback(func() { - sd.Destroy() + if err := sd.Destroy(); err != nil { + logger.Warnf("sd.Destroy() = error:%v", err) + } }) for _, v := range sd.GetServices().Values() { diff --git a/registry/file/service_discovery_test.go b/registry/file/service_discovery_test.go index 0bffcae31d..2f52eb52e1 100644 --- a/registry/file/service_discovery_test.go +++ b/registry/file/service_discovery_test.go @@ -44,7 +44,10 @@ func TestNewFileSystemServiceDiscoveryAndDestroy(t *testing.T) { serviceDiscovery, err := newFileSystemServiceDiscovery(testName) assert.NoError(t, err) assert.NotNil(t, serviceDiscovery) - defer serviceDiscovery.Destroy() + defer func () { + err = serviceDiscovery.Destroy() + assert.Nil(t, err) + }() } func TestCURDFileSystemServiceDiscovery(t *testing.T) { @@ -78,8 +81,11 @@ func TestCURDFileSystemServiceDiscovery(t *testing.T) { assert.NoError(t, err) err = serviceDiscovery.Register(r1) - - defer serviceDiscovery.Destroy() + assert.NoError(t, err) + defer func () { + err = serviceDiscovery.Destroy() + assert.NoError(t, err) + }() } func prepareData() { diff --git a/remoting/getty/getty_client_test.go b/remoting/getty/getty_client_test.go index 0b18e973cd..982c509b32 100644 --- a/remoting/getty/getty_client_test.go +++ b/remoting/getty/getty_client_test.go @@ -87,7 +87,9 @@ func getClient(url *common.URL) *Client { exchangeClient := remoting.NewExchangeClient(url, client, 5*time.Second, false) client.SetExchangeClient(exchangeClient) - client.Connect(url) + if err := client.Connect(url); err != nil { + return nil + } return client } @@ -396,7 +398,8 @@ func InitTest(t *testing.T) (*Server, *common.URL) { "side=provider&timeout=3000×tamp=1556509797245&bean.name=UserProvider") // init server userProvider := &UserProvider{} - common.ServiceMap.Register("", url.Protocol, "", "0.0.1", userProvider) + _, err = common.ServiceMap.Register("", url.Protocol, "", "0.0.1", userProvider) + assert.NoError(t, err) invoker := &proxy_factory.ProxyInvoker{ BaseInvoker: *protocol.NewBaseInvoker(url), } diff --git a/remoting/getty/pool.go b/remoting/getty/pool.go index 9689175bcf..c70aeea516 100644 --- a/remoting/getty/pool.go +++ b/remoting/getty/pool.go @@ -149,13 +149,23 @@ func (c *gettyRPCClient) newSession(session getty.Session) error { panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn())) } - tcpConn.SetNoDelay(conf.GettySessionParam.TcpNoDelay) - tcpConn.SetKeepAlive(conf.GettySessionParam.TcpKeepAlive) + if err := tcpConn.SetNoDelay(conf.GettySessionParam.TcpNoDelay); err != nil { + logger.Error("tcpConn.SetNoDelay() = error:%v", err) + } + if err := tcpConn.SetKeepAlive(conf.GettySessionParam.TcpKeepAlive); err != nil { + logger.Error("tcpConn.SetKeepAlive() = error:%v", err) + } if conf.GettySessionParam.TcpKeepAlive { - tcpConn.SetKeepAlivePeriod(conf.GettySessionParam.keepAlivePeriod) + if err := tcpConn.SetKeepAlivePeriod(conf.GettySessionParam.keepAlivePeriod); err != nil { + logger.Error("tcpConn.SetKeepAlivePeriod() = error:%v", err) + } + } + if err := tcpConn.SetReadBuffer(conf.GettySessionParam.TcpRBufSize); err != nil { + logger.Error("tcpConn.SetReadBuffer() = error:%v", err) + } + if err := tcpConn.SetWriteBuffer(conf.GettySessionParam.TcpWBufSize); err != nil { + logger.Error("tcpConn.SetWriteBuffer() = error:%v", err) } - tcpConn.SetReadBuffer(conf.GettySessionParam.TcpRBufSize) - tcpConn.SetWriteBuffer(conf.GettySessionParam.TcpWBufSize) session.SetName(conf.GettySessionParam.SessionName) session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen) diff --git a/remoting/zookeeper/client_test.go b/remoting/zookeeper/client_test.go index 34741700ca..af09edba48 100644 --- a/remoting/zookeeper/client_test.go +++ b/remoting/zookeeper/client_test.go @@ -81,7 +81,10 @@ func verifyEventStateOrder(t *testing.T, c <-chan zk.Event, expectedStates []zk. func Test_newMockZookeeperClient(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.Nil(t, err) + }() states := []zk.State{zk.StateConnecting, zk.StateConnected, zk.StateHasSession} verifyEventStateOrder(t, event, states, "event channel") @@ -92,7 +95,10 @@ func Test_newMockZookeeperClient(t *testing.T) { func TestCreate(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.Nil(t, err) + }() err = z.Create("test1/test2/test3/test4") assert.NoError(t, err) @@ -103,7 +109,10 @@ func TestCreate(t *testing.T) { func TestCreateDelete(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.Nil(t, err) + }() states := []zk.State{zk.StateConnecting, zk.StateConnected, zk.StateHasSession} verifyEventStateOrder(t, event, states, "event channel") @@ -117,7 +126,10 @@ func TestCreateDelete(t *testing.T) { func TestRegisterTemp(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.Nil(t, err) + }() err = z.Create("/test1/test2/test3") assert.NoError(t, err) @@ -131,7 +143,10 @@ func TestRegisterTemp(t *testing.T) { func TestRegisterTempSeq(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.Nil(t, err) + }() err = z.Create("/test1/test2/test3") assert.NoError(t, err) tmpath, err := z.RegisterTempSeq("/test1/test2/test3", []byte("test")) diff --git a/remoting/zookeeper/facade_test.go b/remoting/zookeeper/facade_test.go index 3d5798c947..4a76a80ef1 100644 --- a/remoting/zookeeper/facade_test.go +++ b/remoting/zookeeper/facade_test.go @@ -88,7 +88,11 @@ func (r *mockFacade) IsAvailable() bool { func Test_Facade(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) - defer ts.Stop() + defer func() { + if err := ts.Stop(); err != nil { + t.Errorf("tc.Stop() = error: %v", err) + } + }() url, _ := common.NewURL("mock://127.0.0.1") mock := newMockFacade(z, url) go HandleClientRestart(mock) diff --git a/remoting/zookeeper/listener_test.go b/remoting/zookeeper/listener_test.go index 37ef1b4b96..07976dc31d 100644 --- a/remoting/zookeeper/listener_test.go +++ b/remoting/zookeeper/listener_test.go @@ -95,7 +95,11 @@ func TestListener(t *testing.T) { ` var wait sync.WaitGroup ts, client, event := initZkData(t) - defer ts.Stop() + defer func () { + if err := ts.Stop(); err != nil { + t.Errorf("ts.Stop() = error: %v", err) + } + }() client.Wait.Add(1) wait.Add(1) go client.HandleZkEvent(event)