Leader events no longer pointers

Author: ekoutanov

README.md

@@ -86,14 +86,14 @@ config := Config{
 // wrap up in-flight work upon loss of leader status.
 barrier := func(e Event) {
   switch e.(type) {
-  case *LeaderAcquired:
+  case LeaderAcquired:
     // The application may initialise any state necessary to perform work as a leader.
     log.Infof("Received event: leader elected")
-  case *LeaderRevoked:
+  case LeaderRevoked:
     // The application may block the Barrier callback until it wraps up any in-flight
     // activity. Only upon returning from the callback, will a new leader be elected.
     log.Infof("Received event: leader revoked")
-  case *LeaderFenced:
+  case LeaderFenced:
     // The application must immediately terminate any ongoing activity, on the assumption
     // that another leader may be imminently elected. Unlike the handling of LeaderRevoked,
     // blocking in the Barrier callback will not prevent a new leader from being elected.

goneli_doc_test.go

@@ -92,14 +92,14 @@ func Example_lowLevel() {
 	// wrap up in-flight work upon loss of leader status.
 	barrier := func(e Event) {
 		switch e.(type) {
-		case *LeaderAcquired:
+		case LeaderAcquired:
 			// The application may initialise any state necessary to perform work as a leader.
 			log.Infof("Received event: leader elected")
-		case *LeaderRevoked:
+		case LeaderRevoked:
 			// The application may block the Barrier callback until it wraps up any in-flight
 			// activity. Only upon returning from the callback, will a new leader be elected.
 			log.Infof("Received event: leader revoked")
-		case *LeaderFenced:
+		case LeaderFenced:
 			// The application must immediately terminate any ongoing activity, on the assumption
 			// that another leader may be imminently elected. Unlike the handling of LeaderRevoked,
 			// blocking in the Barrier callback will not prevent a new leader from being elected.

int/neli_int_test.go

@@ -74,13 +74,13 @@ func test(t *testing.T, numNodes int, spawnInterval time.Duration) {
 		scr.I()("Starting Neli %d/%d", i+1, numNodes)
 		n, err := New(config, func(e Event) {
 			switch e.(type) {
-			case *LeaderAcquired:
+			case LeaderAcquired:
 				scr.I()("Elected leader %s", config.Name)
 				assert.Equal(t, 1, countLeaders(ns, nsMutex))
-			case *LeaderRevoked:
+			case LeaderRevoked:
 				scr.I()("Revoked leader %s", config.Name)
 				assert.Equal(t, 0, countLeaders(ns, nsMutex))
-			case *LeaderFenced:
+			case LeaderFenced:
 				scr.I()("Fenced leader %s", config.Name)
 				assert.Equal(t, 0, countLeaders(ns, nsMutex))
 			default:

mockneli.go

@@ -135,11 +135,11 @@ func (m *mockNeli) tryPulse() bool {
 			m.currentStatus.Set(m.targetStatus.Get())
 			switch m.getTargetStatus() {
 			case mockLeaderStatusAcquired:
-				m.barrier(&LeaderAcquired{})
+				m.barrier(LeaderAcquired{})
 			case mockLeaderStatusRevoked:
-				m.barrier(&LeaderRevoked{})
+				m.barrier(LeaderRevoked{})
 			case mockLeaderStatusFenced:
-				m.barrier(&LeaderFenced{})
+				m.barrier(LeaderFenced{})
 			}
 		}
 	})

mockneli_test.go

@@ -34,20 +34,20 @@ func TestTransition(t *testing.T) {
 	isLeader, err := m.Pulse(10 * time.Second)
 	require.Nil(t, err)
 	require.True(t, isLeader)
-	require.Equal(t, &LeaderAcquired{}, event.Get())
+	require.Equal(t, LeaderAcquired{}, event.Get())
 	require.True(t, m.IsLeader())
 	require.NotEqual(t, time.Unix(0, 0), m.Deadline().Last())
 
 	// Transition away from leader state.
 	m.RevokeLeader()
 	require.True(t, m.IsLeader())
-	require.Equal(t, &LeaderAcquired{}, event.Get())
+	require.Equal(t, LeaderAcquired{}, event.Get())
 
 	m.Deadline().Move(time.Unix(0, 0))
 	isLeader, err = m.Pulse(1 * time.Millisecond)
 	require.Nil(t, err)
 	require.False(t, isLeader)
-	require.Equal(t, &LeaderRevoked{}, event.Get())
+	require.Equal(t, LeaderRevoked{}, event.Get())
 	require.False(t, m.IsLeader())
 
 	// Fenced state.
@@ -56,7 +56,7 @@ func TestTransition(t *testing.T) {
 	isLeader, err = m.Pulse(1 * time.Millisecond)
 	require.Nil(t, err)
 	require.False(t, isLeader)
-	require.Equal(t, &LeaderFenced{}, event.Get())
+	require.Equal(t, LeaderFenced{}, event.Get())
 	require.False(t, m.IsLeader())
 
 	require.Nil(t, m.Close())
@@ -100,7 +100,7 @@ func TestMockBackground(t *testing.T) {
 
 	m.AcquireLeader()
 	wait(t).UntilAsserted(func(t check.Tester) {
-		assert.Equal(t, &LeaderAcquired{}, event.Get())
+		assert.Equal(t, LeaderAcquired{}, event.Get())
 		assert.GreaterOrEqual(t, liveCount.GetInt(), 1)
 	})
 
@@ -128,7 +128,7 @@ func TestMockPulseEventuallyLeader(t *testing.T) {
 	var m MockNeli
 	barrier := func(e Event) {
 		switch e.(type) {
-		case *LeaderFenced:
+		case LeaderFenced:
 			m.AcquireLeader()
 		}
 	}

neli.go

@@ -276,7 +276,7 @@ func (n *neli) tryPulse() (bool, error) {
 					// leadership.
 					n.isLeader.Set(1)
 					n.logger().I()("Resumed leader status (heartbeat received)")
-					n.barrier(&LeaderAcquired{})
+					n.barrier(LeaderAcquired{})
 				}
 			} else {
 				// No messages were received during the last poll.
@@ -287,7 +287,7 @@ func (n *neli) tryPulse() (bool, error) {
 					if elapsed := time.Now().Sub(lastReceived); elapsed > *n.config.ReceiveDeadline {
 						n.logger().I()("Fenced leader (heartbeat timed out)")
 						n.isLeader.Set(0)
-						n.barrier(&LeaderFenced{})
+						n.barrier(LeaderFenced{})
 					}
 				}
 			}
@@ -330,7 +330,7 @@ func onAssigned(n *neli, assigned kafka.AssignedPartitions) {
 		n.isAssigned.Set(1)
 		n.isLeader.Set(1)
 		n.lastReceived.Set(time.Now().UnixNano())
-		n.barrier(&LeaderAcquired{})
+		n.barrier(LeaderAcquired{})
 	}
 }
 
@@ -340,7 +340,7 @@ func onRevoked(n *neli, revoked kafka.RevokedPartitions) {
 		n.logger().I()("Lost leader status")
 		n.isAssigned.Set(0)
 		n.isLeader.Set(0)
-		n.barrier(&LeaderRevoked{})
+		n.barrier(LeaderRevoked{})
 	}
 }
 

neli_test.go

@@ -251,7 +251,7 @@ func TestBasicLeaderElectionAndRevocation(t *testing.T) {
 	m.Reset()
 	wait(t).UntilAsserted(func(t check.Tester) {
 		if assert.Equal(t, 1, b.length()) {
-			_ = b.list()[0].(*LeaderAcquired)
+			_ = b.list()[0].(LeaderAcquired)
 		}
 	})
 	wait(t).UntilAsserted(atLeast(1, onLeaderCnt.GetInt))
@@ -277,7 +277,7 @@ func TestBasicLeaderElectionAndRevocation(t *testing.T) {
 	m.Reset()
 	wait(t).UntilAsserted(func(t check.Tester) {
 		if assert.Equal(t, 2, b.length()) {
-			_ = b.list()[1].(*LeaderRevoked)
+			_ = b.list()[1].(LeaderRevoked)
 		}
 	})
 
@@ -352,7 +352,7 @@ func TestLeaderElectionAndRevocation_timeoutAndReconnect(t *testing.T) {
 	wait(t).UntilAsserted(atLeast(1, onLeaderCnt.GetInt))
 	wait(t).UntilAsserted(func(t check.Tester) {
 		if assert.GreaterOrEqual(t, b.length(), 1) {
-			_ = b.list()[0].(*LeaderAcquired)
+			_ = b.list()[0].(LeaderAcquired)
 		}
 	})
 	wait(t).UntilAsserted(m.ContainsEntries().
@@ -361,7 +361,7 @@ func TestLeaderElectionAndRevocation_timeoutAndReconnect(t *testing.T) {
 		Passes(scribe.Count(1)))
 	wait(t).UntilAsserted(func(t check.Tester) {
 		if assert.GreaterOrEqual(t, b.length(), 2) {
-			_ = b.list()[1].(*LeaderFenced)
+			_ = b.list()[1].(LeaderFenced)
 		}
 	})
 	m.Reset()
@@ -375,7 +375,7 @@ func TestLeaderElectionAndRevocation_timeoutAndReconnect(t *testing.T) {
 		Passes(scribe.Count(1)))
 	wait(t).UntilAsserted(func(t check.Tester) {
 		if assert.GreaterOrEqual(t, b.length(), 3) {
-			_ = b.list()[2].(*LeaderAcquired)
+			_ = b.list()[2].(LeaderAcquired)
 		}
 	})