Moved daemon from cluster package to groupcache package

Author: thrawn01

README.md

@@ -83,28 +83,23 @@ import (
     "log/slog"
 
     "github.com/groupcache/groupcache-go/v3"
-    "github.com/groupcache/groupcache-go/v3/cluster"
-    "github.com/groupcache/groupcache-go/v3/data"
-    "github.com/segmentio/fasthash/fnv1"
+	"github.com/groupcache/groupcache-go/v3/data"
+	"github.com/groupcache/groupcache-go/v3/transport/peer"
 )
 
 func ExampleUsage() {
     ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
     defer cancel()
 
-    // Starts an instance of groupcache with the provided transport
-    d, err := cluster.SpawnDaemon(ctx, "192.168.1.1:8080", groupcache.Options{
-        // If transport is nil, defaults to transport.HttpTransport
-        Transport: nil,
-        HashFn:    fnv1.HashBytes64,
-        Logger:    slog.Default(),
-        Replicas:  50,
-    })
+    // SpawnDaemon is a convenience function which Starts an instance of groupcache 
+	// with the provided transport and listens for groupcache HTTP requests on the address provided.
+    d, err := groupcache.SpawnDaemon(ctx, "192.168.1.1:8080", groupcache.Options{})
     if err != nil {
         log.Fatal("while starting server on 192.168.1.1:8080")
     }
 
-    // Manually set peers, or use some discovery system to identify peers
+    // Manually set peers, or use some discovery system to identify peers.
+    // It is safe to call SetPeers() whenever the peer topology changes
     d.SetPeers(ctx, []peer.Info{
         {
             Address: "192.168.1.1:8080",
@@ -121,8 +116,8 @@ func ExampleUsage() {
     })
 
     // Create a new group cache with a max cache size of 3MB
-    group := d.GroupCache.NewGroup("users", 3000000, groupcache.GetterFunc(
-        func(ctx context.Context, id string, dest groupcache.Sink) error {
+    group := d.NewGroup("users", 3000000, groupcache.GetterFunc(
+        func(ctx context.Context, id string, dest data.Sink) error {
 
             // Returns a protobuf struct `User`
             user, err := fetchUserFromMongo(ctx, id)
@@ -140,7 +135,7 @@ func ExampleUsage() {
     ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*500)
     defer cancel()
 
-    if err := group.Get(ctx, "12345", groupcache.ProtoSink(&user)); err != nil {
+    if err := group.Get(ctx, "12345", data.ProtoSink(&user)); err != nil {
         log.Fatal(err)
     }
 
@@ -155,7 +150,7 @@ func ExampleUsage() {
         log.Fatal(err)
     }
 
-    // Shutdown the daemon
+    // Shutdown the instance and HTTP listeners
     d.Shutdown(ctx)
 }
 ```
@@ -238,6 +233,11 @@ called at any point during `groupcache.Instance` operation as peers leave or joi
 
 If `SetPeers()` is not called, then the `groupcache.Instance` will operate as a local only cache.
 
+### groupcache.Daemon
+This is a convenience struct which encapsulates a `groupcache.Instance` to simplify starting and stopping an instance and
+the associated transport. Calling `groupcache.SpawnDaemon()` calls `SpawnTransport()` on the provided transport to
+listen for incoming requests.
+
 ### data.Group
 Holds the cache that makes up the "group" which can be shared with other instances of group cache. Each
 `groupcache.Instance` must create the same group using the same group name. Group names are how a "group" cache is
@@ -269,17 +269,8 @@ which peer in the cluster is our instance, `Instance.SetPeers()` will return an
 Contains data structures used by groupcache to serialize data between remote instances.
 
 ### cluster package
-Is a convenience package containing functions to easily spawn and shutdown groupcache instances (called daemons) or to
-create a local cluster of group cache instances using the default `transport.HttpTransport`.
-
-**SpawnDaemon()** Spawns a single instance of groupcache using the config provided. The returned *Daemon has methods which
-make interacting with the groupcache instance simple.
-
-```go
-// Starts an instance of groupcache with the provided transport
-d, _ := cluster.SpawnDaemon(ctx, "192.168.1.1:8080", groupcache.Options{})
-defer d.Shutdown(context.Background())
-```
+Is a convenience package containing functions to easily spawn and shutdown a cluster of groupcache instances 
+(called daemons).
 
 **Start()** and **StartWith()** starts a local cluster of groupcache daemons suitable for testing. Users who wish to
 test groupcache in their own project test suites can use these methods to start and stop clusters.

cluster/cluster.go

@@ -16,28 +16,8 @@ limitations under the License.
 
 /*
 Package cluster contains convince functions which make managing the creation of multiple groupcache instances
-simple.
-
-# SpawnDaemon()
-
-Spawns a single instance of groupcache using the config provided. The returned *Daemon has methods which
-make interacting with the groupcache instance simple.
-
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-	defer cancel()
-
-	// Starts an instance of groupcache with the provided transport
-	d, err := cluster.SpawnDaemon(ctx, "192.168.1.1:8080", groupcache.Options{})
-	if err != nil {
-		log.Fatal("while starting server on 192.168.1.1:8080")
-	}
-
-	d.Shutdown(context.Background())
-
-# Start() and StartWith()
-
-Starts a local cluster of groupcache daemons suitable for testing. Users who wish to test groupcache in their
-own project test suites can use these methods to start and stop clusters. See cluster_test.go for more examples.
+simple. To start a local cluster of groupcache daemons suitable for testing you can call cluster.Start()
+or cluster.StartWith(). See cluster_test.go for more examples.
 
 	err := cluster.Start(context.Background(), 2, groupcache.Options{})
 	require.NoError(t, err)
@@ -59,7 +39,7 @@ import (
 	"github.com/groupcache/groupcache-go/v3/transport/peer"
 )
 
-var _daemons []*Daemon
+var _daemons []*groupcache.Daemon
 var _peers []peer.Info
 
 // ListPeers returns a list of all peers in the cluster
@@ -68,12 +48,12 @@ func ListPeers() []peer.Info {
 }
 
 // ListDaemons returns a list of all daemons in the cluster
-func ListDaemons() []*Daemon {
+func ListDaemons() []*groupcache.Daemon {
 	return _daemons
 }
 
 // DaemonAt returns a specific daemon
-func DaemonAt(idx int) *Daemon {
+func DaemonAt(idx int) *groupcache.Daemon {
 	return _daemons[idx]
 }
 
@@ -83,8 +63,12 @@ func PeerAt(idx int) peer.Info {
 }
 
 // FindOwningDaemon finds the daemon which owns the key provided
-func FindOwningDaemon(key string) *Daemon {
-	c, isRemote := _daemons[0].GroupCache.PickPeer(key)
+func FindOwningDaemon(key string) *groupcache.Daemon {
+	if len(_daemons) == 0 {
+		panic("'_daemon' is empty; start a cluster with Start() or StartWith()")
+	}
+
+	c, isRemote := _daemons[0].GetInstance().PickPeer(key)
 	if !isRemote {
 		return _daemons[0]
 	}
@@ -124,7 +108,7 @@ func StartWith(ctx context.Context, peers []peer.Info, opts groupcache.Options)
 	}
 
 	for _, p := range peers {
-		d, err := SpawnDaemon(ctx, p.Address, opts)
+		d, err := groupcache.SpawnDaemon(ctx, p.Address, opts)
 		if err != nil {
 			return fmt.Errorf("StartWith: while starting daemon for '%s': %w", p.Address, err)
 		}
@@ -158,7 +142,7 @@ func Restart(ctx context.Context) error {
 		if err := _daemons[i].Start(ctx); err != nil {
 			return err
 		}
-		_ = _daemons[i].GroupCache.SetPeers(ctx, _peers)
+		_ = _daemons[i].GetInstance().SetPeers(ctx, _peers)
 	}
 	return nil
 }

cluster/daemon.go

@@ -0,0 +1,111 @@
+/*
+Copyright Derrick J Wippler
+
+Licensed 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.
+*/
+
+package groupcache
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/groupcache/groupcache-go/v3/data"
+	"github.com/groupcache/groupcache-go/v3/transport"
+	"github.com/groupcache/groupcache-go/v3/transport/peer"
+)
+
+// Daemon is an instance of groupcache
+type Daemon struct {
+	instance *Instance
+	opts     Options
+	address  string
+}
+
+// SpawnDaemon starts a new instance of daemon with the config provided
+func SpawnDaemon(ctx context.Context, address string, opts Options) (*Daemon, error) {
+	if opts.Logger == nil {
+		opts.Logger = slog.Default()
+	}
+
+	if opts.Transport == nil {
+		opts.Transport = transport.NewHttpTransport(transport.HttpTransportOptions{})
+	}
+
+	daemon := &Daemon{
+		address: address,
+		opts:    opts,
+	}
+
+	return daemon, daemon.Start(ctx)
+}
+
+func (d *Daemon) Start(ctx context.Context) error {
+	d.instance = New(d.opts)
+	return d.opts.Transport.SpawnServer(ctx, d.address)
+}
+
+// NewGroup is a convenience method which calls NewGroup on the instance associated with this daemon.
+func (d *Daemon) NewGroup(name string, cacheBytes int64, getter Getter) (data.Group, error) {
+	return d.instance.NewGroup(name, cacheBytes, getter)
+}
+
+// GetGroup is a convenience method which calls GetGroup on the instance associated with this daemon
+func (d *Daemon) GetGroup(name string) data.Group {
+	return d.instance.GetGroup(name)
+}
+
+// RemoveGroup is a convenience method which calls RemoveGroup on the instance associated with this daemon
+func (d *Daemon) RemoveGroup(name string) {
+	d.instance.RemoveGroup(name)
+}
+
+// GetInstance returns the current groupcache instance associated with this daemon
+func (d *Daemon) GetInstance() *Instance {
+	return d.instance
+}
+
+// SetPeers is a convenience method which calls SetPeers on the instance associated with this daemon. In
+// addition, it finds and marks this instance as self by asking the transport for it's listening address
+// before calling SetPeers() on the instance. If this is not desirable, call Daemon.GetInstance().SetPeers()
+// instead.
+func (d *Daemon) SetPeers(ctx context.Context, src []peer.Info) error {
+	dest := make([]peer.Info, len(src))
+	for idx := 0; idx < len(src); idx++ {
+		dest[idx] = src[idx]
+		if dest[idx].Address == d.ListenAddress() {
+			dest[idx].IsSelf = true
+		}
+	}
+	return d.instance.SetPeers(ctx, dest)
+}
+
+// MustClient is a convenience method which creates a new client for this instance. This method will
+// panic if transport.NewClient() returns an error.
+func (d *Daemon) MustClient() peer.Client {
+	c, err := d.opts.Transport.NewClient(context.Background(), peer.Info{Address: d.ListenAddress()})
+	if err != nil {
+		panic(err)
+	}
+	return c
+}
+
+// ListenAddress returns the address this instance is listening on
+func (d *Daemon) ListenAddress() string {
+	return d.opts.Transport.ListenAddress()
+}
+
+// Shutdown attempts a clean shutdown of the daemon and all related resources.
+func (d *Daemon) Shutdown(ctx context.Context) error {
+	return d.opts.Transport.ShutdownServer(ctx)
+}

daemon.go

@@ -21,6 +21,8 @@ import (
 	"time"
 )
 
+// TODO: Attempt to move this into `transport` package and Sink back into `groupcache` package
+
 type Group interface {
 	Set(context.Context, string, []byte, time.Time, bool) error
 	Get(context.Context, string, Sink) error