Flutter'da WebSocket ile Gerçek Zamanlı Uygulamalar

Flutter'da WebSocket ile Gerçek Zamanlı Uygulamalar

REST API, istek-cevap modeliyle çalışır; gerçek zamanlı güncellemeler için ise sunucudan anlık veri akışı gerekir. WebSocket, tek bir TCP bağlantısı üzerinden çift yönlü veri aktarımı sağlar. Flutter'da hem saf WebSocket API'si hem de socket_io_client ve SignalR gibi üst düzey çözümler kullanılabilir.

WebSocket Bağlantı Yöneticisi

// pubspec: web_socket_channel: ^3.0.1

enum WsStatus { disconnected, connecting, connected, reconnecting }

class WebSocketManager {
  static const _maxReconnectAttempts = 5;
  static const _reconnectDelay = Duration(seconds: 3);

  WebSocketChannel? _channel;
  WsStatus _status = WsStatus.disconnected;
  int _reconnectAttempts = 0;
  Timer? _reconnectTimer;
  Timer? _heartbeatTimer;

  final String _url;
  final String? _token;
  final StreamController<dynamic> _messageController =
      StreamController.broadcast();
  final StreamController<WsStatus> _statusController =
      StreamController.broadcast();

  WebSocketManager({required String url, String? token})
      : _url = url, _token = token;

  Stream<dynamic> get messages => _messageController.stream;
  Stream<WsStatus> get status => _statusController.stream;
  WsStatus get currentStatus => _status;

  Future<void> connect() async {
    if (_status == WsStatus.connected || _status == WsStatus.connecting) return;
    _setStatus(WsStatus.connecting);

    try {
      final uri = Uri.parse(_url);
      final headers = _token != null ? {'Authorization': 'Bearer $_token'} : <String, String>{};

      _channel = WebSocketChannel.connect(uri, protocols: ['chat']);
      await _channel!.ready;

      _setStatus(WsStatus.connected);
      _reconnectAttempts = 0;
      _startHeartbeat();

      _channel!.stream.listen(
        _onMessage,
        onError: _onError,
        onDone: _onDisconnect,
        cancelOnError: false,
      );
    } catch (e) {
      debugPrint('WebSocket bağlantı hatası: $e');
      _scheduleReconnect();
    }
  }

  void _onMessage(dynamic data) {
    if (data == 'pong') return; // heartbeat yanıtı
    _messageController.add(data);
  }

  void _onError(dynamic error) {
    debugPrint('WebSocket hatası: $error');
    _setStatus(WsStatus.disconnected);
    _scheduleReconnect();
  }

  void _onDisconnect() {
    _heartbeatTimer?.cancel();
    _setStatus(WsStatus.disconnected);
    _scheduleReconnect();
  }

  void _startHeartbeat() {
    _heartbeatTimer = Timer.periodic(const Duration(seconds: 25), (_) {
      send({'type': 'ping'});
    });
  }

  void _scheduleReconnect() {
    if (_reconnectAttempts >= _maxReconnectAttempts) {
      debugPrint('Maksimum yeniden bağlanma denemesi aşıldı');
      return;
    }
    _reconnectAttempts++;
    _setStatus(WsStatus.reconnecting);
    final delay = _reconnectDelay * _reconnectAttempts; // exponential backoff
    _reconnectTimer = Timer(delay, connect);
  }

  void send(Map<String, dynamic> message) {
    if (_status != WsStatus.connected) return;
    _channel?.sink.add(jsonEncode(message));
  }

  void _setStatus(WsStatus status) {
    _status = status;
    _statusController.add(status);
  }

  void disconnect() {
    _reconnectTimer?.cancel();
    _heartbeatTimer?.cancel();
    _reconnectAttempts = _maxReconnectAttempts; // yeniden bağlanmayı engelle
    _channel?.sink.close();
    _setStatus(WsStatus.disconnected);
  }

  void dispose() {
    disconnect();
    _messageController.close();
    _statusController.close();
  }
}

Chat BLoC ile Mesaj İşleme

@riverpod
class ChatNotifier extends _$ChatNotifier {
  late WebSocketManager _ws;
  StreamSubscription? _sub;

  @override
  ChatState build(String roomId) {
    _ws = WebSocketManager(
      url: 'wss://api.example.com/ws/rooms/$roomId',
      token: ref.read(authProvider).token,
    );
    _ws.connect();

    _sub = _ws.messages.listen(_handleMessage);
    ref.onDispose(() {
      _sub?.cancel();
      _ws.dispose();
    });

    return const ChatState();
  }

  void _handleMessage(dynamic raw) {
    final data = jsonDecode(raw as String) as Map<String, dynamic>;
    switch (data['type']) {
      case 'message':
        final msg = ChatMessage.fromJson(data['payload'] as Map<String, dynamic>);
        state = state.copyWith(messages: [msg, ...state.messages]);
      case 'typing':
        final userId = data['userId'] as String;
        state = state.copyWith(
          typingUserIds: data['isTyping'] as bool
              ? {...state.typingUserIds, userId}
              : state.typingUserIds.where((id) => id != userId).toSet(),
        );
      case 'presence':
        final userId = data['userId'] as String;
        final online = data['online'] as bool;
        state = state.copyWith(
          onlineUserIds: online
              ? {...state.onlineUserIds, userId}
              : state.onlineUserIds.where((id) => id != userId).toSet(),
        );
      case 'message_read':
        final msgId = data['messageId'] as String;
        state = state.copyWith(
          messages: state.messages.map((m) =>
              m.id == msgId ? m.copyWith(isRead: true) : m).toList(),
        );
    }
  }

  void sendMessage(String text) {
    final tempId = const Uuid().v4();
    final msg = ChatMessage(
      id: tempId, text: text, senderId: ref.read(currentUserProvider).id,
      createdAt: DateTime.now(), status: MessageStatus.sending,
    );
    state = state.copyWith(messages: [msg, ...state.messages]);

    _ws.send({
      'type': 'message',
      'payload': {'text': text, 'tempId': tempId},
    });
  }

  Timer? _typingTimer;
  void setTyping(bool isTyping) {
    _ws.send({'type': 'typing', 'isTyping': isTyping});
    if (isTyping) {
      _typingTimer?.cancel();
      _typingTimer = Timer(const Duration(seconds: 3), () => setTyping(false));
    }
  }
}

Bağlantı Durumu Göstergesi

class ConnectionStatusBar extends ConsumerWidget {
  const ConnectionStatusBar({super.key});

  @override
  Widget build(BuildContext context, WidgetRef ref) {
    return StreamBuilder<WsStatus>(
      stream: ref.read(wsManagerProvider).status,
      builder: (context, snapshot) {
        final status = snapshot.data ?? WsStatus.disconnected;

        return AnimatedContainer(
          duration: const Duration(milliseconds: 300),
          height: status == WsStatus.connected ? 0 : 28,
          color: switch (status) {
            WsStatus.connected    => Colors.green,
            WsStatus.connecting   => Colors.orange,
            WsStatus.reconnecting => Colors.orange[700],
            WsStatus.disconnected => Colors.red,
          },
          child: Center(
            child: Text(
              switch (status) {
                WsStatus.connected    => 'Bağlandı',
                WsStatus.connecting   => 'Bağlanıyor...',
                WsStatus.reconnecting => 'Yeniden bağlanıyor...',
                WsStatus.disconnected => 'Bağlantı yok',
              },
              style: const TextStyle(color: Colors.white, fontSize: 12),
            ),
          ),
        );
      },
    );
  }
}

// Yazıyor göstergesi
class TypingIndicator extends StatelessWidget {
  final Set<String> typingUserIds;
  const TypingIndicator({super.key, required this.typingUserIds});

  @override
  Widget build(BuildContext context) {
    if (typingUserIds.isEmpty) return const SizedBox.shrink();
    return Padding(
      padding: const EdgeInsets.symmetric(horizontal: 16, vertical: 4),
      child: Row(
        children: [
          const _BouncingDots(),
          const SizedBox(width: 8),
          Text(
            typingUserIds.length == 1
                ? '${typingUserIds.first} yazıyor...'
                : '${typingUserIds.length} kişi yazıyor...',
            style: TextStyle(color: Colors.grey[600], fontSize: 12),
          ),
        ],
      ),
    );
  }
}

WebSocket uygulamalarında en kritik konu güvenilir yeniden bağlanma stratejisidir. Exponential backoff ile yeniden bağlanma, heartbeat ile bağlantı sağlığı kontrolü ve uygulama arka plana geçtiğinde bağlantıyı yönetmek, üretim kalitesinde bir gerçek zamanlı deneyim için gereklidir.