Software caused connection abort

Hi, I have the following problem

I have a chatbot that connects through eventsource to receive messages, the issue is that sometimes when the user switches app and comes back, it throws the following error:

{
"type": "error",
"message": "Software caused connection abort",
"xhrStatus": 200,
"xhrState": 4
}

And the application crashes, stops sending messages and I can't connect again until I restart the app.

Hi, this looks like a normal error event to me. The app shouldn't crash from an error event like this. Are you maybe throwing an exception in the error handler?

Also check out this comment about handling reconnections when the app is resumed after being suspended: #65 (comment)

Yes, this is how I handle the reconnections, but as I said, from time to time it works and in others it throws that error and stops working, I will leave the implementation I have.

// chatcontext.tsx

import { getClient } from '@/services/botpress.services';
import { useTextToSpeech } from '@/hooks/useTextToSpeech';
import { useAuthStore } from '@/hooks/useAuthStore';
 
export const ChatProvider = ({ children }: ChatProviderProps) => {
  const user = useAuthStore((state) => state.user);
  const countries = userSoa?.cltId;
  const clientRef = useRef(getClient());
  const [isTyping, setIsTyping] = useState(false);
  const appState = useRef(AppState.currentState);
  const { hasVoice } = useVoiceStore();
  const { speech } = useTextToSpeech();
  const [appStateVisible, setAppStateVisible] = useState(appState.current);

  const {
    userId,
    userToken,
    setUserId,
    setUserToken,
    addMessage,
    setConversationId,
    conversationId,
    messages,
  } = useChatStore();
  const userData = {
    fullname: userSoa?.first_name!,
    email: user?.email!,
    countries: countries?.map((country) => CLIENTS[country]).join(', ')!,
  };

  useEffect(() => {
    const subscription = AppState.addEventListener('change', (nextAppState) => {
      if (appState.current.match(/inactive|background/) && nextAppState === 'active') {
        console.log('La aplicación ha vuelto al primer plano!');
        clientRef.current.connect();
      } else if (nextAppState === 'background' || nextAppState === 'inactive') {
        console.log('La aplicación ha ido al fondo!');
        clientRef.current.disconnect();
      }
      appState.current = nextAppState;
      setAppStateVisible(appState.current);
    });

    return () => {
      subscription.remove();
    };
  }, []);

  useEffect(() => {
    const client = clientRef.current;

    const connectClient = async () => {
      try {
        if (userId && userToken) {
          await client.connect(
            {
              userId,
              userToken,
            },
            userData,
            {
              name: user?.first_name!,
            },
          );
        } else {
          const res = await client.connect(undefined, userData, {
            name: user?.first_name!,
          });
          setUserId(res?.user.id);
          setUserToken(res?.key);
        }

        if (conversationId) {
          await client.switchConversation(conversationId);
        }
      } catch (err) {
        console.log(err);
      }
    };

    connectClient();

    const handleMessage = (message: any) => {
      addMessage({
        id: message?.id,
        conversationId: message?.conversationId,
        direction: 'incoming',
        authorId: message?.authorId,
        sentOn: message.sentOn,
        payload: message?.payload,
      });
      if (message?.payload?.block?.type === 'text' && hasVoice) {
        console.log('reproduciendo mensaje');
        speech(message?.payload?.block?.text);
      }
    };

    client.on('message', handleMessage);

    const handleTyping = (typing: any) => {
      setIsTyping(typing.isTyping);
    };

    client.on('isTyping', handleTyping);

    return () => {
      client.off('message', handleMessage);
      client.off('isTyping', handleTyping);
      client.disconnect();
    };
  }, [hasVoice, appStateVisible]);

  const sendMessage = async (message: string) => {
    try {
      await clientRef.current.sendMessage({ type: 'text', text: message });
      addMessage({
        direction: 'outgoing',
        sentOn: new Date(),
        text: message,
        type: 'text',
      });
      if (!conversationId) {
        setConversationId(clientRef.current.conversationId);
      }
    } catch (error) {
      console.error('Error al enviar mensaje:', error);
    }
  };
  return (
    <ChatContext.Provider
      value={{
        isTyping,
        messages,
        sendMessage,
      }}>
      {children}
    </ChatContext.Provider>
  );
};

// botpress.services.ts

export const getClient = () => {
  return new Botpress();
};

class Botpress extends EventEmitter {
  public userToken: string = '';
  public userId: string = '';
  public conversationId: string = '';
  private _state: any = {};

  constructor() {
    super();
    this._state = {};
  }

  private async _userExists({ userToken }: UserCredentials) {
    try {
      return await botpressApi
        .get('/users/me', {
          headers: {
            'x-user-key': userToken,
          },
        })
        .then(({ data }) => {
          return data;
        });
    } catch (err) {
      console.log(err);
      return false;
    }
  }

  private async _initialConnect(data: InitialConnect) {
    try {
      return await botpressApi
        .post<any, AxiosResponse<UsersResponse>>('/users', data)
        .then(({ data }) => {
          this.userToken = data.key;
          this.userId = data.user.id;
          return data;
        });
    } catch (err) {
      console.log(err);
    }
  }

  private async _reconnect({ userToken }: UserCredentials, data: InitialConnect) {
    try {
      return await botpressApi
        .put(`/users/me`, data, {
          headers: {
            'x-user-key': userToken,
          },
        })
        .then(({ data }) => {
          this.userToken = userToken;
          this.userId = data.user.id;
          return data;
        });
    } catch (err) {
      console.log(err);
    }
  }

  public async connect(
    userCredentials?: UserCredentials,
    userData: UserData = {} as UserData,
    userOptions: UserOptions = {} as UserOptions,
  ) {
    return userCredentials
      ? (await this._userExists(userCredentials))
        ? this._reconnect(userCredentials, { data: userData, ...userOptions })
        : this._initialConnect({ data: userData, ...userOptions })
      : this._initialConnect({ data: userData, ...userOptions });
  }

  private async _createNewConversation({
    status,
    userId,
    userKey,
  }: {
    userKey: string;
    userId: string;
    status: string;
  }) {
    console.log('Creando nueva conversación...');
    const {
      conversation: { id },
    } = await this.createConversations();
    this.conversationId = id;
    // Connect the user to the conversation
    await this._connectConversation({ userId: this.userId, userToken: userKey }, id);
    return this._createEvent({
      userToken: userKey,
      conversationId: id,
      payload: {
        type: 'conversation_started',
        data: {},
      },
    });
  }

  private async _createEvent({
    userToken,
    conversationId,
    payload,
  }: {
    userToken: string;
    conversationId: string;
    payload: {
      type: string;
      data: any;
    };
  }) {
    try {
      return await botpressApi
        .post(
          `/events`,
          { conversationId, payload },
          {
            headers: {
              'x-user-key': userToken,
            },
          },
        )
        .then(({ data }) => data);
    } catch (err) {
      console.log(err);
    }
  }

  private async createConversations(data = {}) {
    try {
      return await botpressApi
        .post('/conversations', data, {
          headers: {
            'x-user-key': this.userToken,
          },
        })
        .then(({ data }) => data);
    } catch (err) {
      console.log(err);
    }
  }

  private async _connectConversation(userCredentials: UserCredentials, conversationId: string) {
    if (this._state.signalEmitter) {
      console.log('Desconectando el canal existente...');
      this.disconnect();
    }

    const url = `https://webchat.botpress.cloud/${getEnv('EXPO_PUBLIC_BOTPRESS_CLIENT_ID')}/conversations/${conversationId}/listen`;
    const eventSource = new EventSource(url, {
      headers: { 'x-user-key': userCredentials.userToken },
      lineEndingCharacter: '\n',
    });

    // Limpiar listeners previos
    eventSource.removeAllEventListeners();

    // Manejo del evento 'open'
    eventSource.addEventListener('open', () => {
      console.log('Conexión a la conversación abierta');
      this.emit('conversation', conversationId);
      this._state = {
        status: 'conversation_created',
        userId: userCredentials.userId,
        userKey: userCredentials.userToken,
        conversationId: conversationId,
        signalEmitter: eventSource,
      };
    });

    // Manejo del evento 'message'
    eventSource.addEventListener('message', (event) => {
      if (event.data === 'ping') return;
      const payload = event.data ? JSON.parse(event.data) : event;
      try {
        this._handleEvent(payload, userCredentials);
      } catch (error) {
        console.error('Error al procesar el mensaje:', error);
      }
    });

    // Manejo del evento 'error'
    eventSource.addEventListener('error', async (error) => {
      console.error('Error en la conexión del EventStream:', error);
      this.emit('error', new Error(`Conexión a la conversación perdida: ${error.message}`));
    });

    // Guardar el EventSource en el estado
    this._state.signalEmitter = eventSource;
  }


  private _handleEvent(data: any, userCredentials: UserCredentials) {
    const type = data.type;

    switch (type) {
      case 'message_created':
        if (data.data.userId !== userCredentials.userId) {
          this.emit('message', this._mapMessage(data.data));
        }
        break;
      case 'webchat_visibility':
        this.emit('webchatVisibility', data.visibility);
        break;
      case 'webchat_config':
        this.emit('webchatConfig', data.config);
        break;
      case 'typing_started':
        // console.log('Está escribiendo...');
        this.emit('isTyping', { isTyping: true, timeout: data.timeout ?? 5000 });
        break;
      case 'typing_stopped':
        // console.log('Dejó de escribir...');
        this.emit('isTyping', { isTyping: false, timeout: 0 });
        break;
      case 'custom':
        this.emit('customEvent', data.event);
        break;
      case 'message':
        // console.log('Mensaje:', data);
        this.emit('ping', data);
        break;
      default:
        console.debug('Evento desconocido:', data);
    }
  }

  private _mapMessage(message: any) {
    const { metadata } = message,
      { payload } = messageAdapter(message.payload);

    return {
      id: message.id,
      conversationId: message.conversationId,
      authorId: message.userId,
      sentOn: new Date(message.createdAt),
      payload: payload,
      metadata,
    };
  }

  public sendMessage = async (payload: { type: string; text: string }) => {
    this._state =
      this._state === 'conversation_created'
        ? this._state
        : {
            status: 'conversation_creating',
            userId: this.userId,
            userKey: this.userToken,
          };

    this._state.status === 'conversation_creating' &&
      !this.conversationId &&
      (await this._createNewConversation(this._state));

    await this._createMessage({
      conversationId: this.conversationId,
      payload,
    });

    this.emit('messageSent', payload);
  };

  private async _createMessage({
    conversationId,
    payload,
  }: {
    conversationId: string;
    payload: {
      type: string;
      text: string;
    };
  }) {
    try {
      return await botpressApi
        .post(
          `/messages`,
          { conversationId, payload },
          {
            headers: {
              'x-user-key': this.userToken,
            },
          },
        )
        .then(({ data }) => data);
    } catch (err) {
      console.log(err);
    }
  }

  public async switchConversation(conversationId: string) {
    this.conversationId = conversationId;
    this._state =
      (this._state === 'conversation_created' && this.conversationId === conversationId) ||
      this._state === 'conversation_creating'
        ? {
            status: 'conversation_creating',
            userId: this.userId,
            userKey: this.userToken,
          }
        : this._state;

    await this._connectConversation(
      { userId: this.userId, userToken: this.userToken },
      conversationId,
    );
  }

  public disconnect() {
    if (this._state.signalEmitter) {
      this._state.signalEmitter.removeAllEventListeners();
      this._state.signalEmitter.close();
      this.removeAllListeners();
      this._state.signalEmitter = null; // Limpiar la referencia
    }
  }

}

The only advice I can give you is to pass the debug: true option to the EventSource and then have a look at the log output.

Thanks... What I could notice with the debug is that the connection was not being closed and new connections were being created every time you went back to the app, I corrected this and so far the app has not broken.

Could this have caused a memory leak that caused the app to stop working at some point?

The bug has been mitigated on Android, but on iOS it still occurs.... Is there a difference in how connections are handled?

Navigation Menu

Repository navigation

Software caused connection abort #71

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Software caused connection abort #71

Description

Activity

EmilJunker commented on Nov 26, 2024

Xizted commented on Nov 26, 2024

EmilJunker commented on Nov 26, 2024

Xizted commented on Nov 26, 2024

Xizted commented on Nov 27, 2024

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions