Retrofit 과 Coroutine 사용시 enequeue 내부적 구현

viewModelScope.launch{
    RetrofitService.naverService.getSearchNews(query = keyword, sort = "sim").run{
    	if (isSuccessful){
        	...
        } else {
        	...
        }
    }
}

Retrofit을 사용할 때 Coroutine과 사용하게 되면 enqueue()의 비동기 코드를 실행하는 코드가 없어도 스스로 값을 가져오게 된다.

그래서 바로 Response<>를 사용해 요청 이후의 응답을 바로 받아 사용을 한다.

 

이 동작에 대해 이해하려 한다.

---

enequeue의 내부적 구현

 

 

Retrofit 사용 API Interface 구현체 내부적 구현

이전 글에서 Interface의 내부적 구현에서 HttpServiceMethod.parseAnnotations()에서 아래 isKotlinSuspendFunction 변수를 사용해 SuspendForResponse와 SuspendForBody를 생성해 return 해준다고 하였다.

boolean isKotlinSuspendFunction = requestFactory.isKotlinSuspendFunction;

if (!isKotlinSuspendFunction) {
      return new CallAdapted<>(requestFactory, callFactory, responseConverter, callAdapter);
    } else if (continuationWantsResponse) {
      //noinspection unchecked Kotlin compiler guarantees ReturnT to be Object.
      return (HttpServiceMethod<ResponseT, ReturnT>)
          new SuspendForResponse<>(
              requestFactory,
              callFactory,
              responseConverter,
              (CallAdapter<ResponseT, Call<ResponseT>>) callAdapter);
    } else {
      //noinspection unchecked Kotlin compiler guarantees ReturnT to be Object.
      return (HttpServiceMethod<ResponseT, ReturnT>)
          new SuspendForBody<>(
              requestFactory,
              callFactory,
              responseConverter,
              (CallAdapter<ResponseT, Call<ResponseT>>) callAdapter,
              continuationBodyNullable);
    }

 

 

loadServiceMethod

그리고 이 모든 정보들을 가지고 loadServiceMethod로 돌아가 result에 SuspendForResponse를 저장하고, method 정보와 함께 serviceMethodCache에 저장한다.

ServiceMethod<?> loadServiceMethod(Method method) {
    ServiceMethod<?> result = serviceMethodCache.get(method);
    if (result != null) return result;

    synchronized (serviceMethodCache) {
      result = serviceMethodCache.get(method);
      if (result == null) {
        result = ServiceMethod.parseAnnotations(this, method);
        serviceMethodCache.put(method, result);
      }
    }
    return result;
  }

serviceMethodCache에는 메서드 정보, 그리고 실행에 대한 함수에 대한 정보(SuspendForResponse, callAdapter, callFactory, requestFactory, responseConverter)를 가지고 있다.

 

InvocationHandler

이렇게 가진 정보로 create의 return 값의 Proxy 생성으로 돌아가서 invoke를 수행한다.

new InvocationHandler() {
  private final Platform platform = Platform.get();
  private final Object[] emptyArgs = new Object[0];

  @Override
  public @Nullable Object invoke(Object proxy, Method method, @Nullable Object[] args)
      throws Throwable {
    // If the method is a method from Object then defer to normal invocation.
    if (method.getDeclaringClass() == Object.class) {
      return method.invoke(this, args);
    }
    args = args != null ? args : emptyArgs;
    return platform.isDefaultMethod(method)
        ? platform.invokeDefaultMethod(method, service, proxy, args)
        : loadServiceMethod(method).invoke(args);
  }
});

loadServiceMethod() 메서드의 결과로 SuspendForResponse객체를 가지고, invoke를 수행하게 된다.

이 SuspendForResponse 클래스는 HttpServiceMethod안에 있고,
이때의 invoke는 HttpServiceMethod.java 안의 SuspendForResponse클래스 내부의 invoke() 메서드가 수행된다.

 

HttpServiceMethod.invoke

이때 adapt(call, args)에서 call은 이전에 requestFactory에서 수행했던 Annotaion 파싱, 메서드 정보, callFactory, responseConverter  등의 정보들을 가지고 수행하게 된다.

// HttpServiceMethod

@Override
  final @Nullable ReturnT invoke(Object[] args) {
    Call<ResponseT> call = new OkHttpCall<>(requestFactory, args, callFactory, responseConverter);
    return adapt(call, args);
  }

 

SuspendForResponse.adapt

위의 adapt(call, args)를 통해 위의 adapt가 실행되는데, callAdapter의 adapt(call)을 실행한다.

여기 생성된 call 객체 안에 requestFactory 안에 "GET"이라는 정보, BASE_URL 등 중요 정보가 있다.

// HttpServiceMethod.java 내부의 class SuspendForBody 안의 adapt

protected Object adapt(Call<ResponseT> call, Object[] args) {
      call = callAdapter.adapt(call);

      //noinspection unchecked Checked by reflection inside RequestFactory.
      Continuation<Response<ResponseT>> continuation =
          (Continuation<Response<ResponseT>>) args[args.length - 1];

      // See SuspendForBody for explanation about this try/catch.
      try {
        return KotlinExtensions.awaitResponse(call, continuation);
      } catch (Exception e) {
        return KotlinExtensions.suspendAndThrow(e, continuation);
      }
    }

그리고 위의 awaitResponse()의 인자인 continuation에 Coroutine을 실행할 코드가 있다.

 

Retrofit Call.awaitResponse

이 부분에서 enqueue()를 이미 진행해주고 있기 때문에 "실행"코드 없이 결과 값을 받을 수 있다.

// KotlinExtensions.kt

suspend fun <T> Call<T>.awaitResponse(): Response<T> {
  return suspendCancellableCoroutine { continuation ->
    continuation.invokeOnCancellation {
      cancel()
    }
    enqueue(object : Callback<T> {
      override fun onResponse(call: Call<T>, response: Response<T>) {
        continuation.resume(response)
      }

      override fun onFailure(call: Call<T>, t: Throwable) {
        continuation.resumeWithException(t)
      }
    })
  }
}

 

resumeWithException() 메서드를 통해 내부적으로 resumeWith() 메서드를 실행하고, 결과값을 받고 Result 클래스를 이용해 값을 전달

@SinceKotlin("1.3")
@InlineOnly
public inline fun <T> Continuation<T>.resume(value: T): Unit =
    resumeWith(Result.success(value))

@SinceKotlin("1.3")
@InlineOnly
public inline fun <T> Continuation<T>.resumeWithException(exception: Throwable): Unit =
    resumeWith(Result.failure(exception))

 

 

Coroutine Dispatchers를 지정하지 않는 이유

아래 코드처럼 Dispachers를 신경 쓰고 지정해 주는 사람도 있다.

하지만 이는 불필요한 작업이다.

Retrofit Call의 메서드로 자체적 비동기 호출을 하고, 결과에 따라 콜백을 처리해 준다.

즉, Dispatchers를 지정해 주어도 코루틴의 흐름을 타지 않고, 자바 스레드 풀에서 돌게 된다.

 

CoroutineScope(Dispatchers.IO).launch{
	RetrofitService.naverService.getSearchNews(query = keyword, sort = "sim").run{
    	if (isSuccessful){
        	...
        } else {
        	...
        }
    }
    
    withContext(Dispatchers.Main){
    	//UI SET
    }
}

 

Retrofit.enqueue

Retrofit의 enqueue는 OkHttp의 enqueue를 실행

OkHttp의 call 객체를 생성하여, 그 call 객체를 이용해 enqueue를 실행한다.

// Retrofit.enqueue()

@Override
  public void enqueue(final Callback<T> callback) {
    Objects.requireNonNull(callback, "callback == null");

    okhttp3.Call call;
    Throwable failure;

    synchronized (this) {
      if (executed) throw new IllegalStateException("Already executed.");
      executed = true;

      call = rawCall;
      failure = creationFailure;
      if (call == null && failure == null) {
        try {
          call = rawCall = createRawCall();
        } catch (Throwable t) {
          throwIfFatal(t);
          failure = creationFailure = t;
        }
      }
    }

    if (failure != null) {
      callback.onFailure(this, failure);
      return;
    }

    if (canceled) {
      call.cancel();
    }

    call.enqueue(
        new okhttp3.Callback() {
          @Override
          public void onResponse(okhttp3.Call call, okhttp3.Response rawResponse) {
            Response<T> response;
            try {
              response = parseResponse(rawResponse);
            } catch (Throwable e) {
              throwIfFatal(e);
              callFailure(e);
              return;
            }

            try {
              callback.onResponse(OkHttpCall.this, response);
            } catch (Throwable t) {
              throwIfFatal(t);
              t.printStackTrace(); // TODO this is not great
            }
          }

          @Override
          public void onFailure(okhttp3.Call call, IOException e) {
            callFailure(e);
          }

          private void callFailure(Throwable e) {
            try {
              callback.onFailure(OkHttpCall.this, e);
            } catch (Throwable t) {
              throwIfFatal(t);
              t.printStackTrace(); // TODO this is not great
            }
          }
        });
  }

 

OkHttp.enqueue

client.dispatcher().enqueue()를 실행

void enqueue(Callback responseCallback, boolean forWebSocket) {
  synchronized (this) {
    if (executed) throw new IllegalStateException("Already Executed");
    executed = true;
  }
  client.getDispatcher().enqueue(new AsyncCall(responseCallback, forWebSocket));
}

 

OkHttp.Dispatcher.enqueue

synchronized 블록을 통해 AsyncCall을 처리

void enqueue(AsyncCall call) {
    synchronized (this) {
      readyAsyncCalls.add(call);

      // Mutate the AsyncCall so that it shares the AtomicInteger of an existing running call to
      // the same host.
      if (!call.get().forWebSocket) {
        AsyncCall existingCall = findExistingCallWithHost(call.host());
        if (existingCall != null) call.reuseCallsPerHostFrom(existingCall);
      }
    }
    promoteAndExecute();
  }

 

AsyncCall

AsyncCall은 Runnable을 상속받아 비동기 실행을 돕는다.

final class AsyncCall extends NamedRunnable {
  private final Callback responseCallback;
  private final boolean forWebSocket;
  private AsyncCall(Callback responseCallback, boolean forWebSocket) {
    super("OkHttp %s", originalRequest.urlString());
    this.responseCallback = responseCallback;
    this.forWebSocket = forWebSocket;
  }

 

AsyncCall 클래스 내부에 promoteAndExecute 메서드는 readyAsyncCall을 runningAsyncCall로 승격시켜 주고 executeOn메서드를 통해 실행시켜 준다.

 

executeOn() 메서드에서는 executorService를 통해 인자로 runnable로 받아 스레드를 이용해 실행시켜 준다.
Executor 구현체는 전달받은 작업을 큐에 넣은 후, 사용할 수 있는 스레드가 존재하면, 해당 스레드에 작업을 실행한다.

 

그래서 Dispatchers를 바꿔주며 Coroutine을 사용해도 사실은 아무런 영향을 미치지 못한다고 한다.

안드로이드에서는 Lifecycle에 따라 코루틴을 자동으로 취소해 주는 lifecycleScope를 이용하거나 viewModelScope를 이용하면 된다.

 

Interface
https://rlaudals2374.medium.com/retrofit의-구현체는-어디에-있을까-e04b7952c408
https://seokzoo.tistory.com/4?source=post_page-----964f4b5d0a5d--------------------------------

Retrofit Dispatchers
https://dalinaum.github.io/android/2021/01/28/retrofit-does-not-need-dispachers-io.html