Replace closer with resource package

go.mod

@@ -102,6 +102,7 @@ require (
 	github.com/streadway/quantile v0.0.0-20150917103942-b0c588724d25 // indirect
 	github.com/stretchr/testify v1.6.1
 	github.com/subosito/gotenv v1.2.1-0.20190917103637-de67a6614a4d // indirect
+	github.com/tj/assert v0.0.0-20171129193455-018094318fb0
 	github.com/twotwotwo/sorts v0.0.0-20160814051341-bf5c1f2b8553
 	github.com/uber-go/tally v3.3.13+incompatible
 	github.com/uber/jaeger-client-go v2.25.0+incompatible

src/aggregator/aggregator/map.go

@@ -35,7 +35,7 @@ import (
 	"github.com/m3db/m3/src/metrics/metric/aggregated"
 	"github.com/m3db/m3/src/metrics/metric/unaggregated"
 	"github.com/m3db/m3/src/x/clock"
-	"github.com/m3db/m3/src/x/close"
+	"github.com/m3db/m3/src/x/resource"
 
 	"github.com/uber-go/tally"
 )
@@ -108,7 +108,7 @@ type metricMap struct {
 	firstInsertAt     time.Time
 	rateLimiter       *rate.Limiter
 	runtimeOpts       runtime.Options
-	runtimeOptsCloser close.SimpleCloser
+	runtimeOptsCloser resource.Closer
 	sleepFn           sleepFn
 	metrics           metricMapMetrics
 }

src/aggregator/runtime/options_manager.go

@@ -21,7 +21,7 @@
 package runtime
 
 import (
-	"github.com/m3db/m3/src/x/close"
+	"github.com/m3db/m3/src/x/resource"
 	"github.com/m3db/m3/src/x/watch"
 )
 
@@ -36,7 +36,7 @@ type OptionsManager interface {
 	// RegisterWatcher registers a watcher that watches updates to runtime options.
 	// When an update arrives, the manager will deliver the update to all registered
 	// watchers.
-	RegisterWatcher(l OptionsWatcher) close.SimpleCloser
+	RegisterWatcher(l OptionsWatcher) resource.Closer
 
 	// Close closes the watcher and all descendent watches
 	Close()
@@ -72,7 +72,7 @@ func (w *optionsManager) RuntimeOptions() Options {
 
 func (w *optionsManager) RegisterWatcher(
 	watcher OptionsWatcher,
-) close.SimpleCloser {
+) resource.Closer {
 	_, watch, _ := w.watchable.Watch()
 
 	// The watchable is always initialized so it's okay to do a blocking read.

src/dbnode/client/connection_pool.go

@@ -31,7 +31,7 @@ import (
 
 	"github.com/m3db/m3/src/dbnode/generated/thrift/rpc"
 	"github.com/m3db/m3/src/dbnode/topology"
-	xclose "github.com/m3db/m3/src/x/close"
+	xresource "github.com/m3db/m3/src/x/resource"
 	murmur3 "github.com/m3db/stackmurmur3/v2"
 
 	"github.com/uber-go/tally"
@@ -68,14 +68,14 @@ type connPool struct {
 }
 
 type conn struct {
-	channel xclose.SimpleCloser
+	channel xresource.Closer
 	client  rpc.TChanNode
 }
 
 // NewConnectionFn is a function that creates a connection.
 type NewConnectionFn func(
 	channelName string, addr string, opts Options,
-) (xclose.SimpleCloser, rpc.TChanNode, error)
+) (xresource.Closer, rpc.TChanNode, error)
 
 type healthCheckFn func(client rpc.TChanNode, opts Options) error
 

src/dbnode/client/connection_pool_test.go

@@ -30,7 +30,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/generated/thrift/rpc"
 	"github.com/m3db/m3/src/dbnode/topology"
 	xclock "github.com/m3db/m3/src/x/clock"
-	xclose "github.com/m3db/m3/src/x/close"
+	xresource "github.com/m3db/m3/src/x/resource"
 	"github.com/stretchr/testify/require"
 
 	"github.com/golang/mock/gomock"
@@ -85,7 +85,7 @@ func TestConnectionPoolConnectsAndRetriesConnects(t *testing.T) {
 
 	fn := func(
 		ch string, addr string, opts Options,
-	) (xclose.SimpleCloser, rpc.TChanNode, error) {
+	) (xresource.Closer, rpc.TChanNode, error) {
 		attempt := int(atomic.AddInt32(&attempts, 1))
 		if attempt == 1 {
 			return nil, nil, fmt.Errorf("a connect error")
@@ -237,7 +237,7 @@ func TestConnectionPoolHealthChecks(t *testing.T) {
 
 	fn := func(
 		ch string, addr string, opts Options,
-	) (xclose.SimpleCloser, rpc.TChanNode, error) {
+	) (xresource.Closer, rpc.TChanNode, error) {
 		attempt := atomic.AddInt32(&newConnAttempt, 1)
 		if attempt == 1 {
 			return channelNone, client1, nil

src/dbnode/client/options.go

@@ -38,11 +38,11 @@ import (
 	"github.com/m3db/m3/src/dbnode/storage/index"
 	"github.com/m3db/m3/src/dbnode/topology"
 	"github.com/m3db/m3/src/x/clock"
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/context"
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/pool"
+	xresource "github.com/m3db/m3/src/x/resource"
 	xretry "github.com/m3db/m3/src/x/retry"
 	"github.com/m3db/m3/src/x/sampler"
 	"github.com/m3db/m3/src/x/serialize"
@@ -319,7 +319,7 @@ func NewOptionsForAsyncClusters(opts Options, topoInits []topology.Initializer,
 
 func defaultNewConnectionFn(
 	channelName string, address string, opts Options,
-) (xclose.SimpleCloser, rpc.TChanNode, error) {
+) (xresource.Closer, rpc.TChanNode, error) {
 	channel, err := tchannel.NewChannel(channelName, opts.ChannelOptions())
 	if err != nil {
 		return nil, nil, err

src/dbnode/client/session.go

@@ -48,12 +48,12 @@ import (
 	"github.com/m3db/m3/src/dbnode/x/xpool"
 	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/clock"
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/context"
 	xerrors "github.com/m3db/m3/src/x/errors"
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/pool"
+	xresource "github.com/m3db/m3/src/x/resource"
 	xretry "github.com/m3db/m3/src/x/retry"
 	"github.com/m3db/m3/src/x/sampler"
 	"github.com/m3db/m3/src/x/serialize"
@@ -136,7 +136,7 @@ type sessionState struct {
 type session struct {
 	state                                sessionState
 	opts                                 Options
-	runtimeOptsListenerCloser            xclose.Closer
+	runtimeOptsListenerCloser            xresource.Closer
 	scope                                tally.Scope
 	nowFn                                clock.NowFn
 	log                                  *zap.Logger

src/dbnode/client/session_fetch_high_concurrency_test.go

@@ -34,8 +34,8 @@ import (
 	"github.com/m3db/m3/src/dbnode/sharding"
 	"github.com/m3db/m3/src/dbnode/topology"
 	"github.com/m3db/m3/src/dbnode/ts"
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/ident"
+	xresource "github.com/m3db/m3/src/x/resource"
 	xtime "github.com/m3db/m3/src/x/time"
 
 	"github.com/golang/mock/gomock"
@@ -102,7 +102,7 @@ func TestSessionFetchIDsHighConcurrency(t *testing.T) {
 	// to be able to mock the entire end to end pipeline
 	newConnFn := func(
 		_ string, addr string, _ Options,
-	) (xclose.SimpleCloser, rpc.TChanNode, error) {
+	) (xresource.Closer, rpc.TChanNode, error) {
 		mockClient := rpc.NewMockTChanNode(ctrl)
 		mockClient.EXPECT().Health(gomock.Any()).
 			Return(healthCheckResult, nil).

src/dbnode/digest/fd_digest.go

@@ -25,12 +25,12 @@ import (
 	"hash/adler32"
 	"os"
 
-	xclose "github.com/m3db/m3/src/x/close"
+	xresource "github.com/m3db/m3/src/x/resource"
 )
 
 // FdWithDigest is a container for a file descriptor and the digest for the file contents.
 type FdWithDigest interface {
-	xclose.Closer
+	xresource.ErrCloser
 
 	// Fd returns the file descriptor.
 	Fd() *os.File

src/dbnode/namespace/namespace_runtime_options.go

@@ -23,7 +23,7 @@ package namespace
 import (
 	"sync"
 
-	xclose "github.com/m3db/m3/src/x/close"
+	xresource "github.com/m3db/m3/src/x/resource"
 	"github.com/m3db/m3/src/x/watch"
 )
 
@@ -88,7 +88,7 @@ type RuntimeOptionsManager interface {
 	// RegisterListener registers a listener for updates to runtime options,
 	// it will synchronously call back the listener when this method is called
 	// to deliver the current set of runtime options.
-	RegisterListener(l RuntimeOptionsListener) xclose.SimpleCloser
+	RegisterListener(l RuntimeOptionsListener) xresource.Closer
 
 	// Close closes the watcher and all descendent watches.
 	Close()
@@ -228,7 +228,7 @@ func (w *runtimeOptionsManager) Get() RuntimeOptions {
 
 func (w *runtimeOptionsManager) RegisterListener(
 	listener RuntimeOptionsListener,
-) xclose.SimpleCloser {
+) xresource.Closer {
 	_, watch, _ := w.watchable.Watch()
 
 	// We always initialize the watchable so always read

src/dbnode/namespace/schema_registry.go

@@ -24,8 +24,8 @@ import (
 	"fmt"
 	"sync"
 
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/ident"
+	xresource "github.com/m3db/m3/src/x/resource"
 	xwatch "github.com/m3db/m3/src/x/watch"
 
 	"go.uber.org/zap"
@@ -145,7 +145,7 @@ func (sr *schemaRegistry) getSchemaHistory(nsIDStr string) (SchemaHistory, error
 func (sr *schemaRegistry) RegisterListener(
 	nsID ident.ID,
 	listener SchemaListener,
-) (xclose.SimpleCloser, error) {
+) (xresource.Closer, error) {
 	if !sr.protoEnabled {
 		return nil, nil
 	}

src/dbnode/namespace/types.go

@@ -25,9 +25,9 @@ import (
 
 	"github.com/m3db/m3/src/cluster/client"
 	"github.com/m3db/m3/src/dbnode/retention"
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/instrument"
+	xresource "github.com/m3db/m3/src/x/resource"
 
 	"github.com/gogo/protobuf/proto"
 )
@@ -205,7 +205,7 @@ type SchemaRegistry interface {
 
 	// RegisterListener registers a schema listener for the namespace.
 	// If proto is not enabled, nil, nil is returned
-	RegisterListener(id ident.ID, listener SchemaListener) (xclose.SimpleCloser, error)
+	RegisterListener(id ident.ID, listener SchemaListener) (xresource.Closer, error)
 
 	// Close closes all the listeners.
 	Close()

src/dbnode/network/server/httpjson/cluster/server.go

@@ -28,8 +28,8 @@ import (
 	ns "github.com/m3db/m3/src/dbnode/network/server"
 	"github.com/m3db/m3/src/dbnode/network/server/httpjson"
 	ttcluster "github.com/m3db/m3/src/dbnode/network/server/tchannelthrift/cluster"
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/context"
+	xresource "github.com/m3db/m3/src/x/resource"
 )
 
 type server struct {
@@ -83,6 +83,6 @@ func (s *server) ListenAndServe() (ns.Close, error) {
 
 	return func() {
 		listener.Close()
-		xclose.TryClose(service)
+		xresource.TryClose(service)
 	}, nil
 }

src/dbnode/network/server/tchannelthrift/cluster/server.go

@@ -25,8 +25,8 @@ import (
 	"github.com/m3db/m3/src/dbnode/generated/thrift/rpc"
 	ns "github.com/m3db/m3/src/dbnode/network/server"
 	"github.com/m3db/m3/src/dbnode/network/server/tchannelthrift"
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/context"
+	xresource "github.com/m3db/m3/src/x/resource"
 
 	"github.com/uber/tchannel-go"
 )
@@ -76,6 +76,6 @@ func (s *server) ListenAndServe() (ns.Close, error) {
 
 	return func() {
 		channel.Close()
-		xclose.TryClose(service)
+		xresource.TryClose(service)
 	}, nil
 }

src/dbnode/persist/fs/files.go

@@ -38,7 +38,6 @@ import (
 	"github.com/m3db/m3/src/dbnode/persist/fs/msgpack"
 	"github.com/m3db/m3/src/dbnode/persist/schema"
 	idxpersist "github.com/m3db/m3/src/m3ninx/persist"
-	xclose "github.com/m3db/m3/src/x/close"
 	xerrors "github.com/m3db/m3/src/x/errors"
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/instrument"
@@ -360,17 +359,6 @@ func openFiles(opener fileOpener, fds map[string]**os.File) error {
 	return firstErr
 }
 
-// TODO(xichen): move closeAll to m3x/close.
-func closeAll(closers ...xclose.Closer) error {
-	multiErr := xerrors.NewMultiError()
-	for _, closer := range closers {
-		if err := closer.Close(); err != nil {
-			multiErr = multiErr.Add(err)
-		}
-	}
-	return multiErr.FinalError()
-}
-
 // DeleteFiles delete a set of files, returning all the errors encountered during
 // the deletion process.
 func DeleteFiles(filePaths []string) error {

src/dbnode/persist/fs/persist_manager.go

@@ -35,8 +35,8 @@ import (
 	m3ninxpersist "github.com/m3db/m3/src/m3ninx/persist"
 	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/clock"
-	xclose "github.com/m3db/m3/src/x/close"
 	"github.com/m3db/m3/src/x/instrument"
+	xresource "github.com/m3db/m3/src/x/resource"
 
 	"github.com/pborman/uuid"
 	"github.com/uber-go/tally"
@@ -93,7 +93,7 @@ type persistManager struct {
 
 	metrics persistManagerMetrics
 
-	runtimeOptsListener xclose.SimpleCloser
+	runtimeOptsListener xresource.Closer
 }
 
 type dataPersistManager struct {

src/dbnode/persist/fs/write.go

@@ -36,6 +36,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/ident"
+	xresource "github.com/m3db/m3/src/x/resource"
 	"github.com/m3db/m3/src/x/serialize"
 	xtime "github.com/m3db/m3/src/x/time"
 
@@ -390,7 +391,7 @@ func (w *writer) closeWOIndex() error {
 		return err
 	}
 
-	return closeAll(
+	return xresource.CloseAll(
 		w.infoFdWithDigest,
 		w.indexFdWithDigest,
 		w.summariesFdWithDigest,