radarbase/radar-push-endpoint

RADAR-PushEndpoint

概述和主要用途

RADAR Push Endpoint是一个提供REST接口的服务，用于将基于推送订阅的API连接到Apache Kafka。目前已集成Garmin健康数据推送服务，并支持通过扩展机制添加其他推送服务，适用于RADAR-base生态系统中的数据集成场景。

核心功能和特性

• 暴露REST接口，接收推送订阅服务的数据
• 与Apache Kafka集成，将推送数据转发至Kafka主题
• 支持Garmin健康数据推送集成（活动、睡眠、压力等指标）
• 可配置的用户仓库和认证验证器，支持OAuth2授权
• 模块化设计，支持扩展集成其他推送服务

配置说明

基本配置结构

推送服务集成通过YAML配置文件进行设置，基本结构如下：

yaml
pushIntegration:
  # 推送服务特定配置
  garmin: 
    enabled: true
    userRepositoryClass: "org.radarbase.push.integration.garmin.user.ServiceUserRepository"
  service-n:  # 其他服务配置示例
    enabled: true
    property-xyz: "value"

Garmin集成配置

基础配置

1. 在Garmin开发者门户配置端点
2. 启用Garmin集成并配置用户仓库：

yaml
pushIntegration:
  garmin:
    enabled: true
    userRepositoryClass: org.radarbase.push.integration.garmin.user.ServiceUserRepository

使用GarminServiceUserRepository

若使用GarminServiceUserRepository，需先在管理门户创建OAuth客户端，配置以下属性：

client_id: your_client_id
client_secret: your_client_secret
grant_type: client_credentials
resources: res_restAuthorizer
scope: "SUBJECT.READ", "MEASUREMENT.READ", "SUBJECT.UPDATE", "MEASUREMENT.CREATE"

然后配置YAML属性：

yaml
pushIntegration:
  garmin:
    enabled: true
    userRepositoryClass: org.radarbase.push.integration.garmin.user.GarminServiceUserRepository
    userRepositoryUrl: "http://radar-rest-sources-backend:8090/rest-sources/backend/"  # 根据部署修改
    userRepositoryClientId: "your_client_id"
    userRepositoryClientSecret: "your_client_secret"
    userRepositoryTokenUrl: "http://managementportal-app:8080/oauth/token/"  # 根据部署修改

使用方法

启动服务

使用Docker Compose启动服务：

shell
docker-compose up -d --build

创建Kafka主题

待Kafka就绪后，创建数据接收主题：

shell
TOPIC=test  # 替换为实际主题名
docker-compose exec kafka-1 kafka-topics --create --topic $TOPIC --bootstrap-server kafka-1:9092

访问端点

服务默认通过http://localhost:8090/push/integrations/访问。Garmin具体端点包括：

• 日常数据：http://localhost:8090/push/integrations/garmin/dailies
• 活动数据：http://localhost:8090/push/integrations/garmin/activities
• 活动详情：http://localhost:8090/push/integrations/garmin/activityDetails
• 手动活动：http://localhost:8090/push/integrations/garmin/manualActivities
• 时段数据：http://localhost:8090/push/integrations/garmin/epochs
• 睡眠数据：http://localhost:8090/push/integrations/garmin/sleeps
• 身体成分：http://localhost:8090/push/integrations/garmin/bodyCompositions
• 压力数据：http://localhost:8090/push/integrations/garmin/stress
• 用户指标：http://localhost:8090/push/integrations/garmin/userMetrics
• 运动识别：http://localhost:8090/push/integrations/garmin/moveIQ
• 血氧数据：http://localhost:8090/push/integrations/garmin/pulseOx
• 呼吸数据：http://localhost:8090/push/integrations/garmin/respiration
• 注销端点：http://localhost:8090/push/integrations/garmin/deregister

扩展集成新服务

扩展步骤概述

新推送服务集成需实现于org.radarbase.push.integration.<service-name>包，主要包括以下组件：

1. 资源（Resource）：定义REST端点
2. 用户仓库（UserRepository）：提供用户信息和授权
3. 认证验证器（AuthValidator）：请求认证与授权
4. 转换器（Converter）：数据格式转换（可选）
5. 配置类与资源增强器：服务注册与配置

详细实现步骤

1. 创建资源类

定义服务所需的REST端点，参考GarminPushEndpoint实现。

2. 实现用户仓库

创建实现UserRepository接口的类，提供用户信息查询功能。

3. 实现认证验证器

创建实现radar-jersey库中https://github.com/RADAR-base/radar-jersey/blob/master/src/main/kotlin/org/radarbase/jersey/auth/AuthValidator.kt%E6%8E%A5%E5%8F%A3%E7%9A%84%E7%B1%BB%EF%BC%8C%E5%A4%84%E7%90%86%E8%AF%B7%E6%B1%82%E8%AE%A4%E8%AF%81%E3%80%82

4. 创建资源增强器

注册服务组件到Jersey上下文：

kotlin
class ServiceXIntegrationResourceEnhancer(private val config: Config) : JerseyResourceEnhancer {
    override fun ResourceConfig.enhance() {
        packages(
            "org.radarbase.push.integration.servicex.resource",
            "org.radarbase.push.integration.servicex.filter"
        )
    }

    override fun AbstractBinder.enhance() {
        bind(config.pushIntegration.servicex.userRepository)
            .to(UserRepository::class.java)
            .named("servicex")
            .`in`(Singleton::class.java)

        bind(ServiceXAuthValidator::class.java)
            .to(AuthValidator::class.java)
            .named("servicex")
            .`in`(Singleton::class.java)
    }
}

5. 配置认证委托

在DelegatedAuthValidator中添加服务路由：

kotlin
fun delegate(): AuthValidator {
    return when {
        uriInfo.matches(GARMIN_QUALIFIER) -> namedValidators.named(GARMIN_QUALIFIER).get()
        uriInfo.matches("servicex") -> namedValidators.named("servicex").get()  // 添加新服务路由
        else -> throw IllegalStateException()
    }
}

6. 添加配置类

在Config中添加服务配置：

kotlin
data class PushIntegrationConfig(
    val garmin: GarminConfig = GarminConfig(),
    val servicex: ServiceXConfig  // 新服务配置
)

data class ServiceXConfig(
    val enabled: Boolean = false,
    val userRepositoryClass: String,
    val property1: String,  // 服务特定属性
    val property2: List<String>
)

7. 注册资源增强器

在PushIntegrationEnhancerFactory中注册新服务：

kotlin
if(config.pushIntegration.servicex.enabled) {
    enhancersList.add(ServiceXIntegrationResourceEnhancer(config))
}