fix: add nil reader check in ProcessPendingNotifications to prevent panic

- Add nil check for proto.Reader parameter in both PushNotificationProcessor and VoidPushNotificationProcessor
- Prevent segmentation violation when ProcessPendingNotifications is called with nil reader
- Return early with nil error when reader is nil (graceful handling)
- Fix panic in TestProcessPendingNotificationsEdgeCases test

This addresses the runtime panic that occurred when rd.Buffered() was called on a nil reader,
ensuring robust error handling in edge cases where the reader might not be properly initialized.

Author: ndyakov

pubsub.go

@@ -436,9 +436,9 @@ func (c *PubSub) newMessage(reply interface{}) (interface{}, error) {
 		default:
 			// Try to handle as generic push notification
 			ctx := c.getContext()
-			registry := c.pushProcessor.GetRegistry()
-			if registry != nil {
-				handled := registry.HandleNotification(ctx, reply)
+			handler := c.pushProcessor.GetHandler(kind)
+			if handler != nil {
+				handled := handler.HandlePushNotification(ctx, reply)
 				if handled {
 					// Return a special message type to indicate it was handled
 					return &PushNotificationMessage{

push_notifications.go

@@ -142,6 +142,10 @@ func (p *PushNotificationProcessor) GetRegistryForTesting() *PushNotificationReg
 
 // ProcessPendingNotifications checks for and processes any pending push notifications.
 func (p *PushNotificationProcessor) ProcessPendingNotifications(ctx context.Context, rd *proto.Reader) error {
+	// Check for nil reader
+	if rd == nil {
+		return nil
+	}
 
 	// Check if there are any buffered bytes that might contain push notifications
 	if rd.Buffered() == 0 {
@@ -255,6 +259,11 @@ func (v *VoidPushNotificationProcessor) GetRegistryForTesting() *PushNotificatio
 
 // ProcessPendingNotifications reads and discards any pending push notifications.
 func (v *VoidPushNotificationProcessor) ProcessPendingNotifications(ctx context.Context, rd *proto.Reader) error {
+	// Check for nil reader
+	if rd == nil {
+		return nil
+	}
+
 	// Read and discard any pending push notifications to clean the buffer
 	for {
 		// Peek at the next reply type to see if it's a push notification