I’ve implemented a chat functionality on an app, however, I’m finding it difficult to get real-time chat updates, as well as updating UI with new messages. My Chat gets sent to the database but no real-time updates happening using the pusher_channels_flutter package.

I have initialized pusher, subscribed to the required channel and connected but I don’t get instant updates on the UI on both ends.

ChatController.dart

`import '../model/chat_inbox.dart';
import '../model/chat_model.dart';
import '../model/chat_para.dart';
import '/packages.dart';
// import '/src/features/chat/presentation/views/chat_room_screen.dart';
import '/src/features/chat/repository/chat_repo.dart';

final chatControlerProvider =
    StateNotifierProvider<ChatController, bool>((ref) {
  return ChatController(chatRepository: ref.watch(chatRepoProvider), ref: ref);
});

final allMessageProvider =
    FutureProvider.family<List<ChatInbox>, ChatParas>((ref, model) async {
  return ref
      .watch(chatControlerProvider.notifier)
      .getRoomMessages(model.buyerId, model.sellerId);
});

final allChatMessageProvider = FutureProvider<List<ChatModel>>((ref) async {
  return ref.watch(chatControlerProvider.notifier).getAllMessages();
});

final allChatProvider = StateProvider<List<ChatModel>>((ref) {
  return [];
});


class ChatController extends StateNotifier<bool> {
  ChatRepository _chatRepository;
  Ref _ref;
  ChatController({required ChatRepository chatRepository, required Ref ref})
      : _chatRepository = chatRepository,
        _ref = ref,
        super(false);

  Future<List<ChatModel>> getAllMessages() async {
    final res = await _chatRepository.getAllMessages();
    return res.fold((l) {
      AppConfig.handleErrorMessage(l.error);
      return [];
    }, (allmessage) {
      return allmessage;
    });
  }

  Future<List<ChatInbox>> getRoomMessages(
      String buyerId, String sellerId) async {
    final res = await _chatRepository.getRoomMessages(buyerId, sellerId);
    return res.fold((l) => AppConfig.handleErrorMessage(l.error), (r) => r);
  }

  sendMessage(String content, int buyerId, int receiverId, int adId,
      String? imageUrl, String roomId) async {

    state = true;

    final res = await _chatRepository.sendMessage(
        content, buyerId, receiverId, adId, imageUrl, roomId);
    res.fold((l) {}, (r) => r);

    state = false;
  }
}`

`import 'dart:convert';

import 'package:fpdart/fpdart.dart';
import 'package:marketplaceng/src/features/chat/controller/chat_controller.dart';
import 'package:marketplaceng/src/features/chat/model/chat_inbox.dart';
import 'package:pusher_channels_flutter/pusher_channels_flutter.dart';
import 'package:http/http.dart' as http;

import '../model/chat_model.dart';
import '/packages.dart';

final chatRepoProvider = Provider<ChatRepository>((ref) {
  ChatRepository chatRepo = ChatRepository(
    dioClient: ref.read(dioClientProvider),
    ref: ref,
    localStorage: ref.read(localStorageProvider),
  );

  return chatRepo;
});

ChatRepository.dart

class ChatRepository {
  DioClient _dioClient;
  LocalStorage _localStorage;
  Ref _ref;
  // late WebSocketChannel _channel;
  late PusherChannelsFlutter _pusher;
  PusherChannel? _channel;
  String? _currentChannelName;

  ChatRepository(
      {required DioClient dioClient,
      required LocalStorage localStorage,
      required Ref ref})
      : _localStorage = localStorage,
        _ref = ref,
        _dioClient = dioClient;

  Future<void> initializePusher() async {
    try {
      print('Initializing chat repository');

      String accessToken = await _localStorage.get(Endpoints.access_token);
      _pusher = PusherChannelsFlutter.getInstance();

      Map<String, Map<String, String>>? authParams = {
        'headers': {
          'Authorization': 'Bearer $accessToken',
          'Accept': 'application/json',
          'Content-Type': 'application/x-www-form-urlencoded',
        }
      };

      await _pusher.init(
          apiKey: "xiygjhfdgdu66665488765657665650",
          cluster: "mt1",
          useTLS: true,
          enabledTransports: ['ws', 'wss'],
          authEndpoint: Endpoints.pusherAuthEndpoint,
          authParams: authParams,
          onConnectionStateChange: onConnectionStateChange,
          onError: onError,
          onSubscriptionSucceeded: onSubscriptionSucceeded,
          onEvent: onEvent,
          onSubscriptionError: onSubscriptionError,
          onDecryptionFailure: onDecryptionFailure,
          onMemberAdded: onMemberAdded,
          onMemberRemoved: onMemberRemoved,
          onSubscriptionCount: onSubscriptionCount,
          onAuthorizer: onAuthorizer);
    } catch (e) {
      print('Error initializing chat: $e');
    }
  }

  Future<void> initializeChannel(String roomId) async {
    if (_currentChannelName == roomId && _channel != null) return;

    try {
      await initializePusher();

      _channel = await _pusher.subscribe(channelName: roomId);
      _currentChannelName = roomId;
      await _pusher.connect();
    } catch (e) {
      print('Error initializing channel: $e');
    }
  }

  FutureVoid sendMessage(String content, int buyerId, int receiverId, int adId,
      String? imageUrl, String roomId) async {
    try {
      await initializeChannel('private-$roomId');
      print('The Private Channel --> private-$roomId');

      print('Sending message:');

      String accessToken = await _localStorage.get(Endpoints.access_token);
      await _dioClient.post(
        Endpoints.sendMessage,
        data: {
          'ads_id': adId,
          'content': content,
          'buyer_id': buyerId,
          'reciever': receiverId,
          'image': imageUrl
        },
        options: Options(headers: {'Authorization': 'Bearer $accessToken'}),
      );

      _channel!.trigger(PusherEvent(
        channelName: _channel!.channelName,
        eventName: 'client-NewChatMessage',
        data: json.encode({
          'content': content,
          'buyer_id': buyerId,
          'receiver': receiverId,
          'image': imageUrl
        }),
      ));

      return right('');
    } on DioError catch (e) {
      print(e.requestOptions.data);
      return left(Failure(e));
    }
  }

  FutureEither<List<ChatModel>> getAllMessages() async {
    try {
      String accessToken = await _localStorage.get(Endpoints.access_token);
      final res = await _dioClient.get(
        Endpoints.allMessage,
        options: Options(headers: {'Authorization': 'Bearer $accessToken'}),
      );
      List allChat = res.data['success'];
      return right(allChat.map((e) => ChatModel.fromJson(e)).toList());
    } on DioError catch (e) {
      return left(Failure(e));
    }
  }

  FutureEither<List<ChatInbox>> getRoomMessages(
      String buyerId, String sellerId) async {
    try {
      String accessToken = await _localStorage.get(Endpoints.access_token);
      final res = await _dioClient.get(
        Endpoints.allRoomMessage(buyerId, sellerId),
        options: Options(headers: {'Authorization': 'Bearer $accessToken'}),
      );

      print("The Room Messages are :=> $res");

      List allChat = res.data['messages'];
      return right(allChat.map((e) => ChatInbox.fromJson(e)).toList());
    } on DioError catch (e) {
      return left(Failure(e));
    }
  }

  void closeSocket() {
    try {
      if (_currentChannelName != null) {
        _pusher.unsubscribe(channelName: _currentChannelName!);
        _currentChannelName = null;
        _channel = null;
        _pusher.disconnect();
        print('Socket connection closed');
      }
    } catch (e) {
      print('Error closing socket: $e');
    }
  }

  FutureVoid checkCoin(String agentId) async {
    try {
      print('Checking coin for agent: $agentId');
      String accessToken = await _localStorage.get(Endpoints.access_token);
      final res = await _dioClient.get(
        Endpoints.checkChatCoin(agentId),
        options: Options(headers: {'Authorization': 'Bearer $accessToken'}),
      );
      print(res.data.toString());
      return right(null);
    } on DioError catch (e) {
      print(e.response?.requestOptions.uri);
      return left(Failure(e));
    }
  }

  void onConnectionStateChange(dynamic currentState, dynamic previousState) {
    print("Connection: $currentState");
  }

  void onError(String message, int? code, dynamic e) {
    print("onError: $message code: $code exception: $e");
  }

  void onEvent(PusherEvent event) {
    Map<String, dynamic> jsonMap = json.decode(event.data);
    print(jsonMap);
    if (jsonMap.containsKey('id')) {
      ChatModel newChat = ChatModel.fromJson(jsonMap);
      _ref
          .read(allChatProvider.notifier)
          .update((state) => [...state, newChat]);
    }
  }

  void onSubscriptionSucceeded(String channelName, dynamic data) {
    print("onSubscriptionSucceeded: $channelName data: $data");
    final me = _pusher.getChannel(channelName)?.me;
    print("Me: $me");
  }

  void onSubscriptionError(String message, dynamic e) {
    print("onSubscriptionError: $message Exception: $e");
  }

  void onDecryptionFailure(String event, String reason) {
    print("onDecryptionFailure: $event reason: $reason");
  }

  void onMemberAdded(String channelName, PusherMember member) {
    print("onMemberAdded: $channelName user: $member");
  }

  void onMemberRemoved(String channelName, PusherMember member) {
    print("onMemberRemoved: $channelName user: $member");
  }

  void onSubscriptionCount(String channelName, int subscriptionCount) {
    print(
        "onSubscriptionCount: $channelName subscriptionCount: $subscriptionCount");
  }

  dynamic onAuthorizer(
      String channelName, String socketId, dynamic options) async {
    String accessToken = await _localStorage.get(Endpoints.access_token);
    var authUrl = Endpoints.pusherAuthEndpoint;
    var result = await http.post(
      Uri.parse(authUrl),
      headers: {
        'Authorization': 'Bearer ${accessToken}',
      },
      body: 'socket_id=' + socketId + '&channel_name=' + channelName,
    );
    var json = jsonDecode(result.body);

    print(json);

    return json;
  }
}`