diff --git a/internal/service/component.go b/internal/service/component.go
index 95c2f327..6bfc533c 100644
--- a/internal/service/component.go
+++ b/internal/service/component.go
@@ -2,8 +2,6 @@ package service
 
 import (
 	"context"
-	"fmt"
-	"io"
 	"time"
 
 	"github.com/briandowns/spinner"
@@ -36,69 +34,17 @@ func (e *Component) OperateComponent(ctx *context.Context, request *serviceProto
 		return err
 	}
 
-	var message string
-	var retries = 0
-
-	responseChan := make(chan *serviceProto.OperateServiceResponse)
-	errorChan := make(chan error)
-	go func(ctx context.Context) {
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			default:
-				response, err := stream.Recv()
-				if err != nil {
-					errorChan <- err
-				}
-				responseChan <- response
-			}
-		}
-	}(*requestCtx)
-
-	for {
-		recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout)
-
-		select {
-		case <-recvCtx.Done():
-			spinnerInstance.Stop()
-			cancel()
-			if !util.CanPerformRetry(retries, constant.MaxRetries) {
-				return nil
-			}
-			stream, err = reconnectOperateStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case err := <-errorChan:
-			spinnerInstance.Stop()
-			cancel()
-			if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) {
-				if err != io.EOF {
-					return err
-				} else if err == io.EOF {
-					log.Info(message)
-				}
-				return nil
-			}
-			stream, err = reconnectOperateStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case response := <-responseChan:
-			spinnerInstance.Stop()
-			cancel()
-			if response != nil {
-				message = util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()})
-				logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()})
-				spinnerInstance.Prefix = fmt.Sprintf(" %s  ", message)
-				spinnerInstance.Start()
-				retries = 0
-			}
-		}
+	reconnect := func() (serviceProto.ServiceService_OperateServiceClient, error) {
+		return reconnectOperateStream(client, requestCtx, request, stream)
 	}
+
+	generateResponse := func(response *serviceProto.OperateServiceResponse) string {
+		message := util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()})
+		logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()})
+		return message
+	}
+
+	return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse)
 }
 
 func reconnectOperateStream(client serviceProto.ServiceServiceClient, requestCtx *context.Context, request *serviceProto.OperateServiceRequest, stream serviceProto.ServiceService_OperateServiceClient) (serviceProto.ServiceService_OperateServiceClient, error) {
diff --git a/internal/service/service.go b/internal/service/service.go
index 26de94b7..bd78613f 100644
--- a/internal/service/service.go
+++ b/internal/service/service.go
@@ -40,69 +40,17 @@ func (e *Service) DeployService(ctx *context.Context, request *serviceProto.Depl
 		return err
 	}
 
-	var message string
-	var retries = 0
-
-	responseChan := make(chan *serviceProto.DeployServiceResponse)
-	errorChan := make(chan error)
-	go func(ctx context.Context) {
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			default:
-				response, err := stream.Recv()
-				if err != nil {
-					errorChan <- err
-				}
-				responseChan <- response
-			}
-		}
-	}(*requestCtx)
-
-	for {
-		recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout)
+	reconnect := func() (serviceProto.ServiceService_DeployServiceClient, error) {
+		return reconnectDeployServiceStream(client, requestCtx, request, stream)
+	}
 
-		select {
-		case <-recvCtx.Done():
-			spinnerInstance.Stop()
-			cancel()
-			if !util.CanPerformRetry(retries, constant.MaxRetries) {
-				return nil
-			}
-			stream, err = reconnectDeployServiceStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case err := <-errorChan:
-			spinnerInstance.Stop()
-			cancel()
-			if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) {
-				if err != io.EOF {
-					return err
-				} else if err == io.EOF {
-					log.Info(message)
-				}
-				return nil
-			}
-			stream, err = reconnectDeployServiceStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case response := <-responseChan:
-			spinnerInstance.Stop()
-			cancel()
-			if response != nil {
-				message = util.GenerateResponseMessage(response.GetServiceResponse())
-				logFailedComponentMessagesOnce(response.GetServiceResponse())
-				spinnerInstance.Prefix = fmt.Sprintf(" %s  ", message)
-				spinnerInstance.Start()
-				retries = 0 // Reset retries on successful response
-			}
-		}
+	generateResponse := func(response *serviceProto.DeployServiceResponse) string {
+		message := util.GenerateResponseMessage(response.GetServiceResponse())
+		logFailedComponentMessagesOnce(response.GetServiceResponse())
+		return message
 	}
+
+	return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse)
 }
 
 func logFailedComponentMessagesOnce(response *serviceProto.ServiceResponse) {
@@ -214,74 +162,22 @@ func (e *Service) DeployReleasedService(ctx *context.Context, request *servicePr
 	if err != nil {
 		return err
 	}
+	reconnect := func() (serviceProto.ServiceService_DeployReleasedServiceClient, error) {
+		return reconnectDeployReleasedServiceStream(client, requestCtx, request, stream)
+	}
 
-	var message string
-	var retries = 0
-
-	responseChan := make(chan *serviceProto.DeployReleasedServiceResponse)
-	errorChan := make(chan error)
-	go func(ctx context.Context) {
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			default:
-				response, err := stream.Recv()
-				if err != nil {
-					errorChan <- err
-				}
-				responseChan <- response
-			}
-		}
-	}(*requestCtx)
-	for {
-		recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout)
-
-		select {
-		case <-recvCtx.Done():
-			spinnerInstance.Stop()
-			cancel()
-			if !util.CanPerformRetry(retries, constant.MaxRetries) {
-				return nil
-			}
-			stream, err = reconnectDeployReleasedServiceStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case err := <-errorChan:
-			spinnerInstance.Stop()
-			cancel()
-			if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) {
-				if err != io.EOF {
-					return err
-				} else if err == io.EOF {
-					log.Info(message)
-				}
-				return nil
-			}
-			stream, err = reconnectDeployReleasedServiceStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case response := <-responseChan:
-			spinnerInstance.Stop()
-			cancel()
-			if response != nil {
-				message = response.ServiceResponse.Message
-				message += fmt.Sprintf("\n Service %s %s", response.ServiceResponse.ServiceStatus.ServiceAction, response.ServiceResponse.ServiceStatus)
-				for _, compMessage := range response.ServiceResponse.ComponentsStatus {
-					message += fmt.Sprintf("\n Component %s %s %s", compMessage.ComponentName, compMessage.ComponentAction, compMessage.ComponentStatus)
-				}
-				logFailedComponentMessagesOnce(response.GetServiceResponse())
-				spinnerInstance.Prefix = fmt.Sprintf(" %s  ", message)
-				spinnerInstance.Start()
-				retries = 0 // Reset retries on successful response
-			}
+	generateResponse := func(response *serviceProto.DeployReleasedServiceResponse) string {
+		message := response.ServiceResponse.Message
+		message += fmt.Sprintf("\n Service %s %s", response.ServiceResponse.ServiceStatus.ServiceAction, response.ServiceResponse.ServiceStatus)
+		for _, compMessage := range response.ServiceResponse.ComponentsStatus {
+			message += fmt.Sprintf("\n Component %s %s %s", compMessage.ComponentName, compMessage.ComponentAction, compMessage.ComponentStatus)
 		}
+		logFailedComponentMessagesOnce(response.GetServiceResponse())
+		return message
 	}
 
+	return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse)
+
 }
 
 // UndeployService undeploy service
@@ -349,68 +245,17 @@ func (e *Service) OperateService(ctx *context.Context, request *serviceProto.Ope
 		return err
 	}
 
-	var message string
-	var retries = 0
-
-	responseChan := make(chan *serviceProto.OperateServiceResponse)
-	errorChan := make(chan error)
-	go func(ctx context.Context) {
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			default:
-				response, err := stream.Recv()
-				if err != nil {
-					errorChan <- err
-				}
-				responseChan <- response
-			}
-		}
-	}(*requestCtx)
-	for {
-		recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout)
+	reconnect := func() (serviceProto.ServiceService_OperateServiceClient, error) {
+		return reconnectOperateStream(client, requestCtx, request, stream)
+	}
 
-		select {
-		case <-recvCtx.Done():
-			spinnerInstance.Stop()
-			cancel()
-			if !util.CanPerformRetry(retries, constant.MaxRetries) {
-				return nil
-			}
-			stream, err = reconnectOperateStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case err := <-errorChan:
-			spinnerInstance.Stop()
-			cancel()
-			if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) {
-				if err != io.EOF {
-					return err
-				} else if err == io.EOF {
-					log.Info(message)
-				}
-				return nil
-			}
-			stream, err = reconnectOperateStream(client, requestCtx, request, stream)
-			if err != nil {
-				return nil
-			}
-			retries++
-		case response := <-responseChan:
-			spinnerInstance.Stop()
-			cancel()
-			if response != nil {
-				message = util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()})
-				logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()})
-				spinnerInstance.Prefix = fmt.Sprintf(" %s  ", message)
-				spinnerInstance.Start()
-				retries = 0
-			}
-		}
+	generateResponse := func(response *serviceProto.OperateServiceResponse) string {
+		message := util.GenerateResponseMessageComponentSpecific(response.GetServiceResponse(), []string{request.GetComponentName()})
+		logFailedComponentMessagesOnceForComponents(response.GetServiceResponse(), []string{request.GetComponentName()})
+		return message
 	}
+
+	return handleStreamResponse(stream, requestCtx, spinnerInstance, reconnect, generateResponse)
 }
 
 // ListService deploys service
diff --git a/internal/service/stream_handler.go b/internal/service/stream_handler.go
new file mode 100644
index 00000000..d46115c4
--- /dev/null
+++ b/internal/service/stream_handler.go
@@ -0,0 +1,90 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/briandowns/spinner"
+	"github.com/dream11/odin/pkg/constant"
+	"github.com/dream11/odin/pkg/util"
+	log "github.com/sirupsen/logrus"
+)
+
+// StreamReceiverInterface defines the Recv method that the stream must implement
+type StreamReceiverInterface[R any] interface {
+	Recv() (R, error)
+}
+
+// ReconnectFunc defines a function type for reconnecting the stream
+type ReconnectFunc[S any] func() (S, error)
+
+// GenerateResponse defines a function type for generating a response message
+type GenerateResponse[R any] func(response R) string
+
+// handleStreamResponse handles the stream response, retries on errors, and updates the spinner
+func handleStreamResponse[S StreamReceiverInterface[R], R any](stream S, requestCtx *context.Context, spinnerInstance *spinner.Spinner, reconnect ReconnectFunc[S], generateResponse GenerateResponse[R]) error {
+	var message string
+	var retries = 0
+	errorChan := make(chan error)
+	responseChan := make(chan R)
+
+	go func(ctx context.Context) {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				response, err := stream.Recv()
+				if err != nil {
+					errorChan <- err
+				}
+				responseChan <- response
+			}
+		}
+	}(*requestCtx)
+
+	for {
+		recvCtx, cancel := context.WithTimeout(*requestCtx, constant.Timeout)
+
+		select {
+		case <-recvCtx.Done():
+			spinnerInstance.Stop()
+			cancel()
+			if !util.CanPerformRetry(retries, constant.MaxRetries) {
+				return nil
+			}
+			var er error
+			stream, er = reconnect()
+			if er != nil {
+				return nil
+			}
+			retries++
+		case err := <-errorChan:
+			spinnerInstance.Stop()
+			cancel()
+			if !util.IsRetryable(err) || !util.CanPerformRetry(retries, constant.MaxRetries) {
+				if err != io.EOF {
+					return err
+				} else if err == io.EOF {
+					log.Info(message)
+				}
+				return nil
+			}
+			stream, err = reconnect()
+			if err != nil {
+				return nil
+			}
+			retries++
+		case response := <-responseChan:
+			spinnerInstance.Stop()
+			cancel()
+			if !util.IsNil(response) {
+				message = generateResponse(response)
+				spinnerInstance.Prefix = fmt.Sprintf(" %s  ", message)
+				spinnerInstance.Start()
+				retries = 0
+			}
+		}
+	}
+}
diff --git a/pkg/util/util.go b/pkg/util/util.go
index 59b8b85c..c6ba1c31 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -8,6 +8,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"reflect"
 	"strconv"
 	"strings"
 	"time"
@@ -213,3 +214,14 @@ func CanPerformRetry(
 
 	return true
 }
+
+// IsNil checks if a value is nil
+func IsNil[T any](v T) bool {
+	val := reflect.ValueOf(v)
+	switch val.Kind() {
+	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
+		return val.IsNil()
+	default:
+		return false
+	}
+}