Skip to content
This repository was archived by the owner on Aug 30, 2024. It is now read-only.

feat: Add DialCache for key-based connection caching #391

Merged
merged 7 commits into from
Jul 21, 2021
Merged

feat: Add DialCache for key-based connection caching #391

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
91c4f66
feat: Add DialCache for key-based connection caching
kylecarbs Jul 20, 2021
857a743
Remove DialOptions
kylecarbs Jul 20, 2021
0af7789
Move DialFunc to Dial
kylecarbs Jul 21, 2021
6aa8048
Add WS options to dial
kylecarbs Jul 21, 2021
b0c0167
Requested changes
kylecarbs Jul 21, 2021
1199814
Add comment
kylecarbs Jul 21, 2021
844385e
Fixup
kylecarbs Jul 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Move DialFunc to Dial
  • Loading branch information
@kylecarbs
kylecarbs committed Jul 21, 2021
commit 0af778936b2642f0e6678110d4de69d8b14bccbe
10 changes: 4 additions & 6 deletions wsnet/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,15 @@ import (
)

// dialerFunc is used to reference a dialer returned for caching.
type dialerFunc func(ctx context.Context, key string) (*Dialer, error)
type dialerFunc func() (*Dialer, error)

// DialCache constructs a new DialerCache.
// The cache clears connections that:
// 1. Are older than the TTL and have no active user-created connections.
// 2. Have been closed.
func DialCache(ttl time.Duration, dialer dialerFunc) *DialerCache {
func DialCache(ttl time.Duration) *DialerCache {
dc := &DialerCache{
ttl: ttl,
dialerFunc: dialer,
closed: make(chan struct{}),
flightGroup: &singleflight.Group{},
mut: sync.RWMutex{},
Expand All @@ -30,7 +29,6 @@ func DialCache(ttl time.Duration, dialer dialerFunc) *DialerCache {
}

type DialerCache struct {
dialerFunc dialerFunc
ttl time.Duration
flightGroup *singleflight.Group

Expand Down Expand Up @@ -98,7 +96,7 @@ func (d *DialerCache) evict() {

// Dial returns a Dialer from the cache if one exists with the key provided,
// or dials a new connection using the dialerFunc.
func (d *DialerCache) Dial(ctx context.Context, key string) (*Dialer, bool, error) {
func (d *DialerCache) Dial(ctx context.Context, key string, dialerFunc func() (*Dialer, error)) (*Dialer, bool, error) {
d.mut.RLock()
if dialer, ok := d.dialers[key]; ok {
closed := false
Expand All @@ -119,7 +117,7 @@ func (d *DialerCache) Dial(ctx context.Context, key string) (*Dialer, bool, erro
d.mut.RUnlock()

dialer, err, _ := d.flightGroup.Do(key, func() (interface{}, error) {
dialer, err := d.dialerFunc(ctx, key)
dialer, err := dialerFunc()
if err != nil {
return nil, err
}
Expand Down
30 changes: 15 additions & 15 deletions wsnet/cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,19 +10,23 @@ import (
)

func TestCache(t *testing.T) {
dialFunc := func(connectAddr string) func() (*Dialer, error) {
return func() (*Dialer, error) {
return DialWebsocket(context.Background(), connectAddr, nil)
}
}

t.Run("Caches", func(t *testing.T) {
connectAddr, listenAddr := createDumbBroker(t)
l, err := Listen(context.Background(), slogtest.Make(t, nil), listenAddr, "")
require.NoError(t, err)
defer l.Close()

cache := DialCache(time.Hour, func(ctx context.Context, key string) (*Dialer, error) {
return DialWebsocket(ctx, connectAddr, nil)
})
_, cached, err := cache.Dial(context.Background(), "example")
cache := DialCache(time.Hour)
_, cached, err := cache.Dial(context.Background(), "example", dialFunc(connectAddr))
require.NoError(t, err)
require.Equal(t, cached, false)
_, cached, err = cache.Dial(context.Background(), "example")
_, cached, err = cache.Dial(context.Background(), "example", dialFunc(connectAddr))
require.NoError(t, err)
require.Equal(t, cached, true)
})
Expand All @@ -33,15 +37,13 @@ func TestCache(t *testing.T) {
require.NoError(t, err)
defer l.Close()

cache := DialCache(time.Hour, func(ctx context.Context, key string) (*Dialer, error) {
return DialWebsocket(ctx, connectAddr, nil)
})
cache := DialCache(time.Hour)

conn, cached, err := cache.Dial(context.Background(), "example")
conn, cached, err := cache.Dial(context.Background(), "example", dialFunc(connectAddr))
require.NoError(t, err)
require.Equal(t, cached, false)
require.NoError(t, conn.Close())
_, cached, err = cache.Dial(context.Background(), "example")
_, cached, err = cache.Dial(context.Background(), "example", dialFunc(connectAddr))
require.NoError(t, err)
require.Equal(t, cached, false)
})
Expand All @@ -52,15 +54,13 @@ func TestCache(t *testing.T) {
require.NoError(t, err)
defer l.Close()

cache := DialCache(0, func(ctx context.Context, key string) (*Dialer, error) {
return DialWebsocket(ctx, connectAddr, nil)
})
cache := DialCache(0)

_, cached, err := cache.Dial(context.Background(), "example")
_, cached, err := cache.Dial(context.Background(), "example", dialFunc(connectAddr))
require.NoError(t, err)
require.Equal(t, cached, false)
cache.evict()
_, cached, err = cache.Dial(context.Background(), "example")
_, cached, err = cache.Dial(context.Background(), "example", dialFunc(connectAddr))
require.NoError(t, err)
require.Equal(t, cached, false)
})
Expand Down