<commit_before>#include "dsa_common.h"

#include "responder.h"

#include "module/security_manager.h"

#include "core/session.h"

#include "message/request/invoke_request_message.h"
#include "message/request/list_request_message.h"
#include "message/request/set_request_message.h"
#include "message/request/subscribe_request_message.h"
#include "stream/responder/outgoing_invoke_stream.h"
#include "stream/responder/outgoing_list_stream.h"
#include "stream/responder/outgoing_set_stream.h"
#include "stream/responder/outgoing_subscribe_stream.h"
#include "stream/simple_stream.h"

namespace dsa {

Responder::Responder(Session &session) : _session(session) {}

void Responder::destroy_impl() {
  for (auto &it : _outgoing_streams) {
    it.second->destroy();
  }
  _outgoing_streams.clear();
}

void Responder::connection_changed() {
  for (auto it = _outgoing_streams.begin(); it != _outgoing_streams.end();) {
    if (it->second->connection_changed()) {
      it = _outgoing_streams.erase(it);
    } else {
      ++it;
    }
  }
}

inline ref_<OutgoingSubscribeStream> Responder::on_subscribe_request(
    ref_<SubscribeRequestMessage> &&message) {
  return make_ref_<OutgoingSubscribeStream>(
      _session.get_ref(), message->get_target_path(), message->get_rid(),
      message->get_subscribe_options());
}

inline ref_<OutgoingListStream> Responder::on_list_request(
    ref_<ListRequestMessage> &&message) {
  return make_ref_<OutgoingListStream>(
      _session.get_ref(), message->get_target_path(), message->get_rid(),
      message->get_list_options());
}

inline ref_<OutgoingInvokeStream> Responder::on_invoke_request(
    ref_<InvokeRequestMessage> &&message) {
  return make_ref_<OutgoingInvokeStream>(
      _session.get_ref(), message->get_target_path(), message->get_rid(),
      std::move(message));
}

inline ref_<OutgoingSetStream> Responder::on_set_request(
    ref_<SetRequestMessage> &&message) {
  return make_ref_<OutgoingSetStream>(_session.get_ref(),
                                      message->get_target_path(),
                                      message->get_rid(), std::move(message));
}

void Responder::receive_message(ref_<Message> &&message) {
  auto find_stream = _outgoing_streams.find(message->get_rid());
  if (find_stream != _outgoing_streams.end()) {
    if (message->type() == MessageType::CLOSE_REQUEST) {
      find_stream->second->destroy();
      _outgoing_streams.erase(find_stream);
    } else {
      find_stream->second->receive_message(std::move(message));
    }
    return;
  }
  if (message->type() == MessageType::CLOSE_REQUEST) {
    // no need to close a stream that doesn't exist
    return;
  }
  auto request = DOWN_CAST<RequestMessage *>(message.get());
  if (request->get_target_path().is_invalid()) {
    MessageType response_type = Message::get_response_type(request->type());
    if (response_type != MessageType::INVALID) {
      _session.write_stream(make_ref_<SimpleStream>(
          request->get_rid(), response_type, MessageStatus::INVALID_MESSAGE));
    }
    return;
  }

  std::function<void(PermissionLevel)> callback;

  switch (message->type()) {
    case MessageType::INVOKE_REQUEST: {
      auto stream =
          on_invoke_request(ref_<InvokeRequestMessage>(std::move(message)));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::READ) {
          auto response = make_ref_<InvokeResponseMessage>();
          response->set_status(MessageStatus::PERMISSION_DENIED);
          stream->send_response(std::move(response));
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }
    case MessageType::SUBSCRIBE_REQUEST: {
      auto stream = on_subscribe_request(
          ref_<SubscribeRequestMessage>(std::move(message)));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::READ) {
          auto response = make_ref_<SubscribeResponseMessage>();
          response->set_status(MessageStatus::PERMISSION_DENIED);
          stream->send_subscribe_response(std::move(response));
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }
    case MessageType::LIST_REQUEST: {
      auto stream =
          on_list_request(ref_<ListRequestMessage>(std::move(message)));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::LIST) {
          stream->update_response_status(MessageStatus::PERMISSION_DENIED);
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }
    case MessageType::SET_REQUEST: {
      auto stream = on_set_request(ref_<SetRequestMessage>(std::move(message)));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::WRITE) {
          auto response = make_ref_<SetResponseMessage>();
          response->set_status(MessageStatus::PERMISSION_DENIED);
          stream->send_response(std::move(response));
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }

    default:
      return;
  }

  _session._strand->security_manager().check_permission(
      _session.dsid(), request->get_permission_token(), request->type(),
      request->get_target_path(), std::move(callback));
}<commit_msg>fix the error that data used after std::move<commit_after>#include "dsa_common.h"

#include "responder.h"

#include "module/security_manager.h"

#include "core/session.h"

#include "message/request/invoke_request_message.h"
#include "message/request/list_request_message.h"
#include "message/request/set_request_message.h"
#include "message/request/subscribe_request_message.h"
#include "stream/responder/outgoing_invoke_stream.h"
#include "stream/responder/outgoing_list_stream.h"
#include "stream/responder/outgoing_set_stream.h"
#include "stream/responder/outgoing_subscribe_stream.h"
#include "stream/simple_stream.h"

namespace dsa {

Responder::Responder(Session &session) : _session(session) {}

void Responder::destroy_impl() {
  for (auto &it : _outgoing_streams) {
    it.second->destroy();
  }
  _outgoing_streams.clear();
}

void Responder::connection_changed() {
  for (auto it = _outgoing_streams.begin(); it != _outgoing_streams.end();) {
    if (it->second->connection_changed()) {
      it = _outgoing_streams.erase(it);
    } else {
      ++it;
    }
  }
}

inline ref_<OutgoingSubscribeStream> Responder::on_subscribe_request(
    ref_<SubscribeRequestMessage> &&message) {
  return make_ref_<OutgoingSubscribeStream>(
      _session.get_ref(), message->get_target_path(), message->get_rid(),
      message->get_subscribe_options());
}

inline ref_<OutgoingListStream> Responder::on_list_request(
    ref_<ListRequestMessage> &&message) {
  return make_ref_<OutgoingListStream>(
      _session.get_ref(), message->get_target_path(), message->get_rid(),
      message->get_list_options());
}

inline ref_<OutgoingInvokeStream> Responder::on_invoke_request(
    ref_<InvokeRequestMessage> &&message) {
  return make_ref_<OutgoingInvokeStream>(
      _session.get_ref(), message->get_target_path(), message->get_rid(),
      std::move(message));
}

inline ref_<OutgoingSetStream> Responder::on_set_request(
    ref_<SetRequestMessage> &&message) {
  return make_ref_<OutgoingSetStream>(_session.get_ref(),
                                      message->get_target_path(),
                                      message->get_rid(), std::move(message));
}

void Responder::receive_message(ref_<Message> &&message) {
  auto find_stream = _outgoing_streams.find(message->get_rid());
  if (find_stream != _outgoing_streams.end()) {
    if (message->type() == MessageType::CLOSE_REQUEST) {
      find_stream->second->destroy();
      _outgoing_streams.erase(find_stream);
    } else {
      find_stream->second->receive_message(std::move(message));
    }
    return;
  }
  if (message->type() == MessageType::CLOSE_REQUEST) {
    // no need to close a stream that doesn't exist
    return;
  }
  auto request = DOWN_CAST<RequestMessage *>(message.get());
  if (request->get_target_path().is_invalid()) {
    MessageType response_type = Message::get_response_type(request->type());
    if (response_type != MessageType::INVALID) {
      _session.write_stream(make_ref_<SimpleStream>(
          request->get_rid(), response_type, MessageStatus::INVALID_MESSAGE));
    }
    return;
  }

  std::function<void(PermissionLevel)> callback;

  switch (message->type()) {
    case MessageType::INVOKE_REQUEST: {
      auto stream =
          on_invoke_request(ref_<InvokeRequestMessage>(message->get_ref()));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::READ) {
          auto response = make_ref_<InvokeResponseMessage>();
          response->set_status(MessageStatus::PERMISSION_DENIED);
          stream->send_response(std::move(response));
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }
    case MessageType::SUBSCRIBE_REQUEST: {
      auto stream = on_subscribe_request(
          ref_<SubscribeRequestMessage>(message->get_ref()));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::READ) {
          auto response = make_ref_<SubscribeResponseMessage>();
          response->set_status(MessageStatus::PERMISSION_DENIED);
          stream->send_subscribe_response(std::move(response));
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }
    case MessageType::LIST_REQUEST: {
      auto stream =
          on_list_request(ref_<ListRequestMessage>(message->get_ref()));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::LIST) {
          stream->update_response_status(MessageStatus::PERMISSION_DENIED);
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }
    case MessageType::SET_REQUEST: {
      auto stream = on_set_request(ref_<SetRequestMessage>(message->get_ref()));
      callback = [stream, this](PermissionLevel permission_level) mutable {
        // it's possible stream is closed before permission check
        if (stream->is_destroyed()) return;

        stream->allowed_permission = permission_level;

        if (permission_level < PermissionLevel::WRITE) {
          auto response = make_ref_<SetResponseMessage>();
          response->set_status(MessageStatus::PERMISSION_DENIED);
          stream->send_response(std::move(response));
          return;
        } else {
          _session._strand->stream_acceptor().add(std::move(stream));
        }
      };
      _outgoing_streams[stream->rid] = std::move(stream);
      break;
    }

    default:
      return;
  }

  _session._strand->security_manager().check_permission(
      _session.dsid(), request->get_permission_token(), request->type(),
      request->get_target_path(), std::move(callback));
}
{'fraction_non_alphanumeric': 0.12898296018950103, 'fraction_numerical': 0.0, 'mean_word_length': 3.6510305614783225, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 155, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2111405', 'n_tokens_mistral': 3904, 'n_tokens_neox': 3800, 'n_words': 607}