Ktor Client Pipelines – Deep Dive

Interceptor من کجاست 🧐؟

خب، احتمالا اولین چیزی که بعد از مهاجرت از Retrofit و OkHttp باهاش مواجه می‌شین اینه که چطوری می‌تونین Interceptorهای شخصی‌سازی‌شده‌ی خودتون رو بیارین و استفاده کنین. بعضی از این Interceptorها این قابلیت رو دارن که هنوز هم داخل همون ساختار قبلی کار کنن و فقط به OkHttp اضافه بشن، و در نهایت کنار پلاگین‌های Ktor هم قرار بگیرن.

fun provideKtorClient(): HttpClient {
    return HttpClient(OkHttp) {
        engine {
            preconfigured = OkHttpClient.Builder()
                .addInterceptor(SimpleHeaderInterceptor())
                .build()
        }
        ...
    }
}

class SimpleHeaderInterceptor : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response {
        val originalRequest = chain.request()

        val newRequest = originalRequest.newBuilder()
            .addHeader("X-Simple-Interceptor", "HelloFromOkHttp")
            .build()

        return chain.proceed(newRequest)
    }
}

اما برای بعضی از Interceptorها این روش ممکنه جواب نده، و دلیلش هم ساختار متفاوت Ktor هست. پس بیاین قبلش یه نگاه بندازیم ببینیم پشت‌صحنه چه اتفاقی می‌افته 😁

توی Ktor ما یه سری Pipeline داریم که هر کدوم چندتا Phase دارن. کدهای Interceptorهای ما روی این فازها و با یه ترتیب مشخص اجرا می‌شن که جلوتر با هر کدومشون آشنا می‌شیم.

Pipeline اصن چی هست؟

توی Ktor Client هیچ کاری یهویی اتفاق نمی‌افته و هر request/response یه مسیر مشخص رو دنبال می‌کنه که شبیه یه سری لوله (Pipeline) هستن. هر کدوم از این لوله‌ها یه سری نقاط دارن که با یه اسم خاص مشخص شدن و بهشون می‌گیم Phase.

وقتی توی Ktor می نویسیم

client.get<MyDto>("/api")

این Pipeline ها پشت زمینه اجرا میشن

┌────────────────────┐
│ HttpRequestPipeline│
└─────────┬──────────┘
          ↓
┌────────────────────┐
│  HttpSendPipeline  │
└─────────┬──────────┘
          ↓
    🌐 NETWORK 🌐
          ↓
┌────────────────────┐
│HttpResponsePipeline│
└─────────┬──────────┘
          ↓
┌────────────────────┐
│HttpReceivePipeline │
└────────────────────┘

1️⃣ HttpRequestPipeline – اینجا درخواست ساخته میشه

توی این Pipeline هنوز خبری از شبکه نیست و درخواست‌ها تازه دارن شکل می‌گیرن. پلاگین‌هایی مثل ContentNegotiation و ContentEncoding توی این Pipeline دیتا کلاس‌های درخواست رو به JSON تبدیل می‌کنن و بعدش بسته به نیاز، zip هم می‌کنن.

همچنین پلاگین‌هایی مثل HttpTimeout اینجا هستن که زمان طول کشیدن یک درخواست رو اندازه می‌گیرن تا اگه طول کشید، timeout بدن. پلاگینی مثل HttpRequestRetry هم تصمیم می‌گیره درخواست دوباره فرستاده بشه یا نه، و حتی Auth توکن‌های JWT رو به هدر درخواست اضافه می‌کنه.

پس می‌تونیم نتیجه بگیریم که اگه قراره یه پلاگین درخواست‌ها رو دستکاری کنه (Body، Header) یا Chain رو دوباره صدا بزنه، این Pipeline دقیقاً برای همین کار طراحی شده.

خب، حالا بریم سراغ فازهای این Pipeline 😎

این فاز ها به ترتیب اجرا میشن

Before     → آمادهسازی پایه
State      → اضافه شدن رفتار و متادیتا
Transform  → تبدیل مدل به فرم قابل ارسال
Render     → ساخت bytes نهایی
Send       → تحویل به شبکه

1️⃣ Before

📍 قبل از اینکه request جدی شکل بگیره

• مقداردهی های اولیه روی request انجام میشه

• تنظیمات پیش فرض کلاینت روی request اعمال میشه

• بعضی اعتبارسنجی های پایه قبل از ادامه ی مسیر انجام میشن

• اگر از همین اول مشکلی باشه، request میتونه fail بشه

خروجی این فاز:
➡️ یک request خام ولی با تنظیمات پایه

2️⃣ State

📍 اضافه شدن state و رفتار به request

• پلاگینها به request «رفتار» اضافه میکنن

• هدرها، کوکی ها، retry policy، auth و… اینجا attach میشن

• request از حالت صرفاً دادهای خارج میشه و قابل ارسال میشه

خروجی این فاز:
➡️ request کامل شده از نظر متادیتا و state

3️⃣ Transform

📍 تبدیل logical body به body قابل ارسال

• body از مدل منطقی (DTO / Object) تبدیل میشه

• هنوز به bytes نرسیده ولی فرمت مشخص میشه (json، form، …)

• negotiation محتوا اینجا انجام میشه

خروجی این فاز:
➡️ body آمادهی render شدن

4️⃣ Render

📍 تبدیل نهایی body به داده ی قابل ارسال

• body به شکل نهایی (bytes / stream) درمیاد

• request از نظر ساختار کاملاً نهایی میشه

• بعد از این مرحله، body دیگه نباید تغییر کنه

خروجی این فاز:
➡️ request نهایی برای ارسال

5️⃣ Send

📍 لحظه ی تحویل request به engine

• request به HttpSendPipeline سپرده میشه

• timeout، retry نهایی، validation یا تغییرات لحظه ی آخر اعمال میشه

• از این نقطه به بعد وارد network میشیم

خروجی این فاز:
➡️ ارسال واقعی request به شبکه 🌐

مثال ساده:

client.requestPipeline.intercept(HttpRequestPipeline.Transform) {     
    println("Request body is still object: ${context.body}")     
    proceed() 
}

2️⃣ HttpSendPipeline – اینجا request فرستاده میشه

توی این Pipeline خودش دیگه تغییر روی درخواست انجام نمی‌شه و در واقع آخرین فرصت قبل از رفتن به شبکه هست. ولی به جاش کارهایی مثل گرفتن لاگ‌های مورد نیاز، اعمال SSL Pinning روی Engineهای مربوطه (مثل OkHttp، CIO یا Darwin) یا حتی سناریوهای cache کردن انجام می‌شه تا درخواست‌هایی که لازم نیست دوباره فرستاده بشن، skip بشن.

اینجا هم این فاز ها به ترتیب اجرا میشن

Before     → آخرین آمادهسازی
State      → تثبیت وضعیت ارسال
Monitoring → مشاهده و لاگ
Engine     → ارتباط واقعی با شبکه
Receive    → تحویل response

1️⃣ Before

📍 آخرین نقطه قبل از ورود به send

• آخرین آماده سازی ها قبل از ارسال انجام میشه

• request هنوز قابل تغییر جزئیه

• اگر اینجا fail بشه، اصلاً وارد engine نمیشیم

خروجی:
➡️ request آماده ی تحویل به sender

2️⃣ State

📍 قفل شدن وضعیت request برای ارسال

• request وارد حالت «در حال ارسال» میشه

• stateهای لازم برای send attach میشن

• کوکی ها نهایی میشن

خروجی:
➡️ request با state پایدار برای ارسال

3️⃣ Monitoring

📍 مشاهده و ثبت فرآیند ارسال

• قبل و بعد از send، رویدادها ثبت میشن

• برای لاگ، متریک یا دیباگ

• روی رفتار request اثر نمیذاره، فقط observe میکنه

خروجی:
➡️ request بدون تغییر، ولی تحت مانیتورینگ

4️⃣ Engine

📍 ورود به لایه ی شبکه واقعی

• request تحویل engine میشه

• TLS / SSL / socket / connection اینجا اتفاق میافته

• ارتباط واقعی با سرور برقرار میشه

خروجی:
➡️ response خام از شبکه

5️⃣ Receive

📍 تحویل response به pipeline بعدی

• response گرفته شده از engine برگردونده میشه

• کنترل از send خارج میشه

• وارد HttpResponsePipeline میشیم

خروجی:
➡️ response خام برای پردازش بعدی

3️⃣ HttpResponsePipeline – ریسپانس خام

اینجا تازه ریسپانس خام از شبکه میاد و توی این Pipeline، ContentNegotiation ریسپانس رو به DTOها تبدیل می‌کنه. اگه از HttpCallValidator استفاده کرده باشیم، قبل از اینکه ریسپانس serialize بشه، اعتبارسنجی می‌شه و می‌شه بر اساس status code، خطای سرور پرتاب کرد.

همچنین لاگ‌گیری روی ریسپانس‌ها توسط پلاگین Logging انجام می‌شه.

HttpClient {
  HttpResponseValidator {
    validateResponse { response ->
      if (!response.status.isSuccess()) {
        throw ServerException()
      }
    }
  }
}

Receive       → مانیتورینگ 
Parse         → خواندن stream
Transform     → تبدیل به مدل
State         → تثبیت وضعیت
After         → پایان lifecycle

1️⃣ Receive

📍 ثبت و مشاهده ی response

• رویداد دریافت response اعلام میشه

• برای لاگ و مانیتورینگ

• هیچ تغییری روی محتوا انجام نمیشه

خروجی:
➡️ response خام بدون تغییر

2️⃣ Parse

📍 خواندن response از حالت stream

• response از stream شبکه خارج میشه

• body قابل خواندن میشه

• هنوز تبدیل معنایی انجام نشده

خروجی:
➡️ داده ی خام قابل پردازش (bytes / text)

3️⃣ Transform

📍 تبدیل body خام به مدل منطقی

• body به مدل مورد انتظار تبدیل میشه

• negotiation محتوا انجام میشه

• اینجا response معنی دار میشه

خروجی:
➡️ response با body تبدیل شده

4️⃣ State

📍 نهایی سازی وضعیت response

• stateهای response ست میشن

• response آماده ی تحویل به caller میشه

• lifecycle call تکمیل میشه

خروجی:
➡️ response پایدار و نهایی

5️⃣ After

📍 پاکسازی و کارهای انتهایی

• cleanup منابع

• hook های بعد از اتمام call

• آخرین نقطه برای observe response

خروجی:
➡️ پایان چرخه ی response

4️⃣ HttpReceivePipeline – تبدیل به خروجی نهایی

این Pipeline، آخرین مرحله‌ست؛ جایی که ریسپانس پردازش شده تبدیل می‌شه به چیزی که caller واقعاً دریافت می‌کنه. بخشی از پلاگین HttpCache هم توی این Pipeline کار می‌کنه و تصمیم می‌گیره که باید از ریسپانس کش شده استفاده بشه یا نه.

علاوه بر این، پلاگین‌هایی مثل ContentEncoding، Logging و HttpCookies هم یه سری عملیات خودشون رو توی این Pipeline انجام می‌دن.

Before      → تغییرات بنیادین (قبل از state)
State       → اعمال state و encoding
After       → پایان دریافت

1️⃣ Before

📍 اولین برخورد منطقی با response

• کارهایی که باید قبل از هر state یا cache انجام بشن

• response هنوز قابل تغییر محتواییه

• مناسب برای تغییرات بنیادین روی body

خروجی:
➡️ response قابل ادامه ی پردازش

2️⃣ State

📍 اعمال state نهایی روی response

• hookهای مربوط به response اینجا اجرا میشن

• encoding نهایی اعمال میشه

• کوکی ها به روزرسانی میشن

• response برای مصرف آماده میشه

خروجی:
➡️ response با state کامل

3️⃣ After

📍 پایان چرخه ی دریافت

• cleanup نهایی

• پایان lifecycle request/response

• تحویل به caller

خروجی:
➡️ response تحویلی به لایه ی بالاتر

کلاس های Phase و Pipeline

همون‌طور که تو بخش قبل یاد گرفتیم، ساختار داخلی پلاگین‌های Ktor به این شکله که یه‌سری Pipeline داریم که هر کدوم از چند تا فاز تشکیل شدن. فازهای Default اسم‌های مشخصی دارن و با استفاده از همین اسم‌ها می‌تونیم فازهای custom خودمون رو قبل یا بعد از اون‌ها اضافه کنیم. این کار باعث می‌شه پلاگینی که می‌سازیم کمترین تداخل رو با بقیه پلاگین‌ها داشته باشه. برای انجام این کار هم می‌تونیم از دو تا تابع insertPhaseAfter و insertPhaseBefore استفاده کنیم.

class Pipeline<TContext, TSubject> {
    fun intercept(
        phase: PipelinePhase, 
        block: suspend PipelineContext<TSubject, TContext>.() -> Unit,
    )
    fun insertPhaseAfter(ref: PipelinePhase, phase: PipelinePhase)
    fun insertPhaseBefore(ref: PipelinePhase, phase: PipelinePhase)
}

class PipelinePhase(val name: String)

abstract class PipelineContext<TSubject, TContext> {
    val subject: TSubject
    val context: TContext

    suspend fun proceed()
    suspend fun proceedWith(subject: TSubject)
}

حالا توی مرحله‌ی بعدی می‌تونیم به این فازها intercept اضافه کنیم تا وقتی درخواست به اون مرحله رسید، این قطعه کد اجرا بشه. برای این‌که از یه فاز به فاز بعدی منتقل بشیم، باید داخل بدنه‌ی interceptor تابع proceed رو صدا بزنیم. اگه proceed صدا زده نشه، pipeline همون‌جا متوقف می‌شه و دیگه کدهای downstream اجرا نمی‌شن.

نکته‌ی مهم: interceptorهایی که به یه فاز اضافه می‌شن، به ترتیبی که intercept شدن اجرا می‌شن. به همین خاطر ترتیب صدا زدن install(plugin)ها خیلی مهمه و می‌تونه روی نتیجه‌ی نهایی تأثیر بذاره.

Pipeline
 ├── Phase 1
 │     ├── interceptor A
 │     ├── interceptor B
 │
 ├── Phase 2
 │     ├── interceptor C
 │
 └── Phase 3
       ├── interceptor D

در ادامه، یه مثال واقعی از دو حالت استفاده از فازهای Custom و Default می‌بینیم.

اینجا توی پلاگین DefaultRequest که base URL رو مشخص کردیم، در واقع داریم به RequestPipeline و فاز Before یه intercept اضافه می‌کنیم. این intercept برای همه‌ی درخواست‌ها اجرا می‌شه و وظیفه‌ش اینه که base URL رو به درخواست اضافه کنه. نکته‌ی مهمش هم اینه که این کار قبل از هر فاز دیگه و عملاً قبل از اجرای هر پایپ‌لاین انجام می‌شه.

val client = HttpClient {
  defaultRequest {
    contentType(ContentType.Application.Json)
    url("https://example.com")
  }
}

// Internal code
public class DefaultRequest private constructor(private val block: DefaultRequestBuilder.() -> Unit) {
    public companion object Plugin : HttpClientPlugin<DefaultRequestBuilder, DefaultRequest> {
        override val key: AttributeKey<DefaultRequest> = AttributeKey("DefaultRequest")

        override fun prepare(block: DefaultRequestBuilder.() -> Unit): DefaultRequest = DefaultRequest(block)

        override fun install(plugin: DefaultRequest, scope: HttpClient) {
            scope.requestPipeline.intercept(HttpRequestPipeline.Before) {
                // logic
            }
        }
    }
}

توی این مثال، پلاگین HttpCallValidator یه کاری که می‌کنه اینه که توی یکی از بخش‌هاش یه فاز سفارشی با اسم BeforeReceive می‌سازه و اون رو قبل از HttpResponsePipeline.Receive اضافه می‌کنه. در واقع این فاز، اولین فازی هست که بعد از رسیدن response اجرا می‌شه. بعد از اینکه کل Pipeline از این فاز به بعد اجرا شد، اگه توی این مسیر خطایی اتفاق بیفته، callback مربوط به handleResponseExceptionWithRequest صدا زده می‌شه.

HttpClient {
  HttpResponseValidator {
    handleResponseExceptionWithRequest { request, cause ->
      if (cause is ResponseException) {
        val response = cause.response
        val status = response.status

        when (status.value) {
          401 -> {
            throw AuthExpiredException(url = request.url.encodedPath, statusCode = status.value)
          }

          403 -> {
            throw AccessDeniedException(message = "Access denied to ${request.url.encodedPath}")
          }

          in 500..599 -> {
            throw ServerException(code = status.value, endpoint = request.url.encodedPath)
          }

          else -> {
            throw cause
          }
        }
      }

      throw cause
    }
  }
}

// Internal code
// ... → BeforeReceive → Receive → Parse → Transform → ...
internal object ReceiveError : ClientHook<suspend (HttpRequest, Throwable) -> Throwable?> {
  override fun install(client: HttpClient, handler: suspend (HttpRequest, Throwable) -> Throwable?) {
    val BeforeReceive = PipelinePhase("BeforeReceive")
    client.responsePipeline.insertPhaseBefore(HttpResponsePipeline.Receive, BeforeReceive)
    client.responsePipeline.intercept(BeforeReceive) {
      try {
        proceed()
      } catch (cause: Throwable) {
        val error = handler(context.request, cause)
        if (error != null) throw error
      }
    }
  }
}

🔩 ClientHook چیه و چرا اصلاً وجود داره؟

توی مثال قبلی دیدیم که پلاگین HttpCallValidator از چیزی به اسم ClientHook ارث‌بری می‌کنه تا بتونه توسط پلاگین HttpResponseValidator روی ResponsePipeline اجرا بشه. در واقع پلاگین‌ها به‌صورت مستقیم از ساختار Pipeline خبر ندارن و فقط با مفهوم Hook کار می‌کنن. این کار عمداً انجام شده تا یه لایه‌ی انتزاعی بین پلاگین‌ها و ساختار داخلی Ktor وجود داشته باشه و از یه‌سری مشکل جلوگیری بشه، مثل این‌که:

• پلاگین به ساختار داخلی Ktor وابسته نشه

• با تغییر pipeline یا phaseها، پلاگین نشکنه

• پلاگین‌ها با هم تداخل نداشته باشن

• یه API عمومی و پایدار در اختیارمون باشه

• بشه از Hookهای مشترک بین پلاگین‌های مختلف استفاده کرد

به‌عنوان مثال، وقتی از SetupRequest استفاده می‌کنیم، عملاً یه همچین روندی پشت‌صحنه در جریانه.

val CustomPlugin = createClientPlugin("CustomPlugin") {
  on(SetupRequest) {
    // Logic
  }
}

public interface ClientHook<HookHandler> {
  public fun install(client: HttpClient, handler: HookHandler)
}

public object SetupRequest : ClientHook<suspend (HttpRequestBuilder) -> Unit>{
  override fun install(
    client: HttpClient,
    handler: suspend (HttpRequestBuilder) -> Unit,
  ) {
    client.requestPipeline.intercept(HttpRequestPipeline.Before) {
      handler(context)
    }
  }
}

🧱 ClientPluginBuilder چیه؟

برای ساختن یه پلاگین Custom، وقتی میایم همچین کدی می‌نویسیم، در واقع پشت‌صحنه داریم از یه کلاسی استفاده می‌کنیم که وظیفه‌ش نگه‌داری اسم پلاگین، کانفیگ‌هایی که از بیرون مقداردهی می‌شن و همین‌طور Hookهاست.

توی مثال پایین هم از تابع onRequest همین کلاس استفاده شده که پشت‌صحنه یه RequestHook به این پلاگین اضافه می‌کنه. بعد از اینکه پلاگین نصب شد، تابع install مربوط به هوک صدا زده می‌شه تا یه Interceptor به فاز State از RequestPipeline اضافه کنه.

از بقیه توابع مهم این کلاس هم می‌شه به onResponse، transformRequestBody و transformResponseBody اشاره کرد.

val CustomPlugin = createClientPlugin("CustomPlugin") {
  onRequest {
    // Logic
  }
}

class ClientPluginBuilder(...) {
  internal val hooks: MutableList<HookHandler<*>> = mutableListOf()

  public fun onRequest(
    block: suspend OnRequestContext.(request: HttpRequestBuilder,
    content: Any) -> Unit,
  ) {
    on(RequestHook, block)
  }

  public fun <HookHandler> on(
    hook: ClientHook<HookHandler>,
    handler: HookHandler,
  ) {
    hooks.add(HookHandler(hook, handler))
  }
}

internal class HookHandler<T>(
  private val hook: ClientHook<T>,
  private val handler: T,
) {
  fun install(client: HttpClient) {
    hook.install(client, handler)
  }
}

internal object RequestHook : ClientHook<suspend OnRequestContext.(request: HttpRequestBuilder, content: Any) -> Unit> {
  override fun install(
    client: HttpClient,
    handler: suspend OnRequestContext.(request: HttpRequestBuilder, content: Any) -> Unit
  ) {
    client.requestPipeline.intercept(HttpRequestPipeline.State) {
      handler(OnRequestContext(), context, subject)
    }
  }
}

به صورت کلی میشه نحوه اتصال پلاگین تا Pipeline به این شکل ترسیم کرد.

Plugin DSL
    ↓
ClientPluginBuilder
    ↓
on(ClientHook)
    ↓
Hook.install()
    ↓
Pipeline.intercept(Phase)

🔍 Hookهای معروف Ktor دقیقاً کجا اجرا میشن؟

جمع بندی نهایی

توی این مقاله سعی کردم چیزهایی که پشت‌صحنه‌ی Ktor اتفاق می‌افته رو توضیح بدم. برخلاف OkHttp که معماری نسبتاً ساده‌ای داره، Ktor از یه معماری مخصوص به خودش استفاده می‌کنه؛ معماری‌ای که در نگاه اول شاید خیلی ساده به نظر بیاد، اما وقتی وارد نوشتن پلاگین‌های Custom می‌شی و می‌بینی پلاگینی که نوشتی داره روی بقیه پلاگین‌ها اثر می‌ذاره، اون‌وقته که باید بری تا تهش رو دربیاری و بفهمی این کتابخونه دقیقاً قراره چه کارهایی انجام بده.

پایان...