fix: fix wrong metric recording and cleanup code base (#440)

Author: Tochemey

actors/actor_system_test.go

@@ -1013,15 +1013,16 @@ func TestActorSystem(t *testing.T) {
 		got := &metricdata.ResourceMetrics{}
 		err = r.Collect(ctx, got)
 		require.NoError(t, err)
-		assert.Len(t, got.ScopeMetrics, 1)
-		assert.Len(t, got.ScopeMetrics[0].Metrics, 5)
+		require.Len(t, got.ScopeMetrics, 1)
+		require.Len(t, got.ScopeMetrics[0].Metrics, 6)
 
 		expected := []string{
 			"actor_child_count",
 			"actor_stash_count",
 			"actor_restart_count",
 			"actors_count",
 			"actor_processed_count",
+			"actor_received_duration",
 		}
 		// sort the array
 		sort.Strings(expected)

actors/api.go

@@ -48,30 +48,22 @@ func Ask(ctx context.Context, to *PID, message proto.Message, timeout time.Durat
 		return nil, ErrDead
 	}
 
-	var messageContext *ReceiveContext
-
-	switch msg := message.(type) {
-	case *internalpb.RemoteMessage:
-		var actual proto.Message
-		if actual, err = msg.GetMessage().UnmarshalNew(); err != nil {
-			return nil, ErrInvalidRemoteMessage(err)
-		}
-		messageContext = newReceiveContext(ctx, NoSender, to, actual).WithRemoteSender(msg.GetSender())
-	default:
-		messageContext = newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender)
+	receiveContext, err := toReceiveContext(ctx, to, message)
+	if err != nil {
+		return nil, err
 	}
 
-	to.doReceive(messageContext)
+	to.doReceive(receiveContext)
 
 	// await patiently to receive the response from the actor
 	select {
-	case response = <-messageContext.response:
-		to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
+	case response = <-receiveContext.response:
+		to.recordLatestReceiveDurationMetric(ctx)
 		return
 	case <-time.After(timeout):
-		to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
+		to.recordLatestReceiveDurationMetric(ctx)
 		err = ErrRequestTimeout
-		to.toDeadletterQueue(messageContext, err)
+		to.toDeadletterQueue(receiveContext, err)
 		return
 	}
 }
@@ -82,66 +74,41 @@ func Tell(ctx context.Context, to *PID, message proto.Message) error {
 		return ErrDead
 	}
 
-	var messageContext *ReceiveContext
-
-	switch msg := message.(type) {
-	case *internalpb.RemoteMessage:
-		var (
-			actual proto.Message
-			err    error
-		)
-
-		if actual, err = msg.GetMessage().UnmarshalNew(); err != nil {
-			return ErrInvalidRemoteMessage(err)
-		}
-		messageContext = newReceiveContext(ctx, NoSender, to, actual).WithRemoteSender(msg.GetSender())
-	default:
-		messageContext = newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender)
+	receiveContext, err := toReceiveContext(ctx, to, message)
+	if err != nil {
+		return err
 	}
 
-	to.doReceive(messageContext)
-	to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
+	to.doReceive(receiveContext)
+	to.recordLatestReceiveDurationMetric(ctx)
 	return nil
 }
 
 // BatchTell sends bulk asynchronous messages to an actor
+// The messages will be processed one after the other in the order they are sent
+// This is a design choice to follow the simple principle of one message at a time processing by actors.
 func BatchTell(ctx context.Context, to *PID, messages ...proto.Message) error {
-	if !to.IsRunning() {
-		return ErrDead
-	}
-
-	for i := 0; i < len(messages); i++ {
-		message := messages[i]
-		messageContext := newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender)
-		to.doReceive(messageContext)
+	// messages are processed one after the other
+	for _, mesage := range messages {
+		if err := Tell(ctx, to, mesage); err != nil {
+			return err
+		}
 	}
-	to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 	return nil
 }
 
 // BatchAsk sends a synchronous bunch of messages to the given PID and expect responses in the same order as the messages.
 // The messages will be processed one after the other in the order they are sent
 // This is a design choice to follow the simple principle of one message at a time processing by actors.
 func BatchAsk(ctx context.Context, to *PID, timeout time.Duration, messages ...proto.Message) (responses chan proto.Message, err error) {
-	if !to.IsRunning() {
-		return nil, ErrDead
-	}
-
 	responses = make(chan proto.Message, len(messages))
 	defer close(responses)
-
-	for i := 0; i < len(messages); i++ {
-		receiveContext := newReceiveContext(ctx, NoSender, to, messages[i])
-		to.doReceive(receiveContext)
-		select {
-		case result := <-receiveContext.response:
-			responses <- result
-			to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
-		case <-time.After(timeout):
-			to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
-			to.toDeadletterQueue(receiveContext, ErrRequestTimeout)
-			return nil, ErrRequestTimeout
+	for _, mesage := range messages {
+		response, err := Ask(ctx, to, mesage, timeout)
+		if err != nil {
+			return nil, err
 		}
+		responses <- response
 	}
 	return
 }
@@ -499,3 +466,17 @@ func RemoteSpawn(ctx context.Context, host string, port int, name, actorType str
 	}
 	return nil
 }
+
+// toReceiveContext creates a ReceiveContext provided a message and a receiver
+func toReceiveContext(ctx context.Context, to *PID, message proto.Message) (*ReceiveContext, error) {
+	switch msg := message.(type) {
+	case *internalpb.RemoteMessage:
+		actual, err := msg.GetMessage().UnmarshalNew()
+		if err != nil {
+			return nil, ErrInvalidRemoteMessage(err)
+		}
+		return newReceiveContext(ctx, NoSender, to, actual).WithRemoteSender(msg.GetSender()), nil
+	default:
+		return newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender), nil
+	}
+}