gRPC と Protocol Buffers — 高速な内部通信

// proto/notification/v1/notification_service.proto
syntax = "proto3";
 
package notification.v1;
 
// 通知サービスの定義
service NotificationService {
  // プッシュ通知を送る
  rpc SendPushNotification(SendPushNotificationRequest)
    returns (SendPushNotificationResponse);
 
  // 複数ユーザーへの一括通知
  rpc BroadcastNotification(BroadcastNotificationRequest)
    returns (BroadcastNotificationResponse);
 
  // 通知履歴の取得（サーバーサイドストリーミング）
  rpc StreamNotifications(StreamNotificationsRequest)
    returns (stream Notification);
}
 
message SendPushNotificationRequest {
  string user_id = 1;
  string title = 2;
  string body = 3;
  map<string, string> data = 4;
}
 
message SendPushNotificationResponse {
  bool success = 1;
  string message_id = 2;
  string error = 3;
}
 
message Notification {
  string id = 1;
  string title = 2;
  string body = 3;
  bool read = 4;
  int64 created_at = 5;  // Unix timestamp
}
 
message StreamNotificationsRequest {
  string user_id = 1;
  int32 limit = 2;
}

// proto/recommendation/v1/recommendation_service.proto
syntax = "proto3";
 
package recommendation.v1;
 
service RecommendationService {
  rpc GetRecommendedProperties(GetRecommendedPropertiesRequest)
    returns (GetRecommendedPropertiesResponse);
}
 
message GetRecommendedPropertiesRequest {
  string user_id = 1;
  int32 limit = 2;
  repeated string exclude_property_ids = 3;
}
 
message GetRecommendedPropertiesResponse {
  repeated RecommendedProperty properties = 1;
}
 
message RecommendedProperty {
  string property_id = 1;
  float score = 2;
  string reason = 3;
}

# Gemfile
gem 'gruf'
gem 'google-protobuf'
gem 'grpc'

# config/initializers/gruf.rb
Gruf.configure do |c|
  c.server_binding_url = '0.0.0.0:9001'
 
  # 認証インターセプター
  c.interceptors.use(Gruf::Authentication::Basic, {
    credentials: [{ username: 'grpc', password: ENV['GRPC_PASSWORD'] }]
  })
 
  # ロギング
  c.interceptors.use(Gruf::Interceptors::Instrumentation::StatsdInterceptor, {
    send_io_timings: true
  })
end

# app/rpc/notification_service_controller.rb
class NotificationServiceController < Gruf::Controllers::Base
  bind Notification::V1::NotificationService::Service
 
  def send_push_notification
    user_id = request.message.user_id
    user = User.find(user_id)
 
    # FCMへのプッシュ通知送信
    result = PushNotificationService.send(
      token: user.fcm_token,
      title: request.message.title,
      body: request.message.body,
      data: request.message.data.to_h
    )
 
    Notification::V1::SendPushNotificationResponse.new(
      success: result.success?,
      message_id: result.message_id,
      error: result.error_message.to_s
    )
  rescue ActiveRecord::RecordNotFound => e
    raise Gruf::Error.new(
      code: :not_found,
      app_code: :USER_NOT_FOUND,
      message: "User #{user_id} not found"
    )
  end
 
  def broadcast_notification
    user_ids = request.message.user_ids
    users = User.where(id: user_ids)
 
    # バックグラウンドジョブで非同期送信
    BroadcastNotificationJob.perform_later(
      user_ids: users.pluck(:id),
      title: request.message.title,
      body: request.message.body
    )
 
    Notification::V1::BroadcastNotificationResponse.new(
      queued_count: users.count
    )
  end
 
  def stream_notifications
    user_id = request.message.user_id
    notifications = Notification.where(user_id: user_id)
                                .order(created_at: :desc)
                                .limit(request.message.limit)
 
    # ストリーミングでレスポンスを返す
    notifications.each do |notification|
      request.active_call.output_stream.push(
        Notification::V1::Notification.new(
          id: notification.id.to_s,
          title: notification.title,
          body: notification.body,
          read: notification.read?,
          created_at: notification.created_at.to_i
        )
      )
    end
  end
end

# app/services/notification_client.rb
class NotificationClient
  STUB_CLASS = Notification::V1::NotificationService::Stub
 
  def initialize
    @stub = STUB_CLASS.new(
      ENV['NOTIFICATION_SERVICE_URL'],  # 例：notification-service:9001
      credentials
    )
  end
 
  def send_push_notification(user_id:, title:, body:, data: {})
    request = Notification::V1::SendPushNotificationRequest.new(
      user_id: user_id.to_s,
      title: title,
      body: body,
      data: data
    )
 
    response = @stub.send_push_notification(request, deadline: Time.now + 5.seconds)
 
    if response.success
      { success: true, message_id: response.message_id }
    else
      { success: false, error: response.error }
    end
  rescue GRPC::NotFound => e
    Rails.logger.error "User not found: #{e.message}"
    { success: false, error: 'user_not_found' }
  rescue GRPC::DeadlineExceeded
    Rails.logger.error 'Notification service timeout'
    { success: false, error: 'timeout' }
  end
 
  private
 
  def credentials
    if Rails.env.production?
      GRPC::Core::ChannelCredentials.new(ssl_cert)
    else
      :this_channel_is_insecure
    end
  end
 
  def ssl_cert
    File.read(Rails.root.join('certs', 'ca.crt'))
  end
end

# コントローラからの使用例
class Api::V1::PropertiesController < Api::V1::BaseController
  def create
    @property = current_user.properties.create!(property_params)
 
    # 物件作成後にオーナーへ通知（gRPCで非同期に近い形で）
    NotificationClient.new.send_push_notification(
      user_id: current_user.id,
      title: '物件を登録しました',
      body: "「#{@property.name}」の審査を開始しました"
    )
 
    render json: PropertySerializer.new(@property).serializable_hash, status: :created
  end
end

// ECSのサービスディスカバリ設定（タスク定義）
{
  "serviceDiscoveryArn": "arn:aws:servicediscovery:...",
  "registries": [
    {
      "registryArn": "arn:aws:servicediscovery:ap-northeast-1:xxx:service/srv-xxx",
      "port": 9001
    }
  ]
}

# docker-compose.yml（開発環境）
services:
  rails-api:
    build: .
    environment:
      - NOTIFICATION_SERVICE_URL=notification-service:9001
      - RECOMMENDATION_SERVICE_URL=recommendation-service:9002
    depends_on:
      - notification-service
 
  notification-service:
    build: ./services/notification
    ports:
      - "9001:9001"
 
  recommendation-service:
    build: ./services/recommendation
    ports:
      - "9002:9002"

# grpc-tools のインストール
gem install grpc-tools
 
# Rubyコードの生成
grpc_tools_ruby_protoc \
  -I proto \
  --ruby_out=lib/proto \
  --grpc_out=lib/proto \
  proto/notification/v1/notification_service.proto

# lib/proto/notification/v1/notification_service_pb.rb（生成コード）
# 手動編集禁止
 
module Notification
  module V1
    SendPushNotificationRequest = Google::Protobuf::DescriptorPool.generated_pool.lookup(
      "notification.v1.SendPushNotificationRequest"
    ).msgclass
    # ...
  end
end

# app/interceptors/logging_interceptor.rb
class LoggingInterceptor < Gruf::Interceptors::ServerInterceptor
  def call
    start_time = Time.now
 
    begin
      result = yield
      duration = ((Time.now - start_time) * 1000).round(2)
 
      Rails.logger.info({
        grpc_method: request.method_key,
        duration_ms: duration,
        status: 'success'
      }.to_json)
 
      result
    rescue Gruf::Error => e
      Rails.logger.error({
        grpc_method: request.method_key,
        error_code: e.code,
        error_message: e.message
      }.to_json)
      raise
    end
  end
end