Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #include <iosfwd>
- #include <iostream>
- #include <sstream>
- #include <boost/iostreams/categories.hpp>
- #include <boost/iostreams/stream.hpp>
- #include <pybind11/pybind11.h>
- #include <pybind11/embed.h>
- namespace py = pybind11;
- class pyobject_source
- {
- public:
- typedef char char_type;
- typedef boost::iostreams::source_tag category;
- pyobject_source(py::object& source_obj)
- : source_obj_(source_obj)
- , pos_(0)
- {
- }
- std::streamsize read(char_type* s, std::streamsize n)
- {
- if (n <= 0) {
- return 0;
- }
- py::object result = source_obj_.attr("read")(n);
- auto const payload = result.cast<std::string>();
- if (payload.empty()) {
- return -1; // EOF
- }
- std::copy(payload.begin(), payload.end(), s);
- return payload.size();
- }
- py::object& source_obj()
- {
- return source_obj_;
- }
- private:
- py::object& source_obj_;
- size_t pos_;
- };
- std::string test_read_stream(std::istream& stream)
- {
- std::ostringstream result;
- std::string line;
- while (std::getline(stream, line)) {
- result << line << '\n';
- }
- return result.str();
- }
- PYBIND11_EMBEDDED_MODULE(testmodule, m)
- {
- m.def("test_read_stream", [](py::object file_like) -> std::string {
- boost::iostreams::stream<pyobject_source> wrapper_stream(file_like);
- return test_read_stream(wrapper_stream);
- });
- }
- int main()
- {
- py::scoped_interpreter guard{};
- try {
- py::exec(R"(\
- import testmodule
- import io
- s = io.BytesIO(b"One\nTwo\nThree")
- print(testmodule.test_read_stream(s))
- )");
- } catch (py::error_already_set& e) {
- std::cerr << e.what() << "\n";
- }
- }
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement