c@90: c@90: #ifndef PIPER_CAPNP_CLIENT_H c@90: #define PIPER_CAPNP_CLIENT_H c@90: c@90: #include "PiperClient.h" c@90: #include "SynchronousTransport.h" c@90: c@90: #include "vamp-support/AssignedPluginHandleMapper.h" c@90: #include "vamp-capnp/VampnProto.h" c@90: c@90: namespace piper { //!!! change c@90: c@90: class PiperCapnpClient : public PiperStubPluginClientInterface c@90: { c@90: // unsigned to avoid undefined behaviour on possible wrap c@90: typedef uint32_t ReqId; c@90: c@90: public: c@90: PiperCapnpClient(SynchronousTransport *transport) : //!!! ownership? shared ptr? c@90: m_transport(transport) { c@90: } c@90: c@90: ~PiperCapnpClient() { c@90: } c@90: c@90: //!!! obviously, factor out all repetitive guff c@90: c@90: //!!! list and load are supposed to be called by application code, c@90: //!!! but the rest are only supposed to be called by the plugin -- c@90: //!!! sort out the api here c@90: c@90: Vamp::Plugin * c@90: load(std::string key, float inputSampleRate, int adapterFlags) { c@90: c@90: if (!m_transport->isOK()) { c@90: throw std::runtime_error("Piper server failed to start"); c@90: } c@90: c@90: Vamp::HostExt::LoadRequest request; c@90: request.pluginKey = key; c@90: request.inputSampleRate = inputSampleRate; c@90: request.adapterFlags = adapterFlags; c@90: c@90: capnp::MallocMessageBuilder message; c@90: RpcRequest::Builder builder = message.initRoot(); c@90: c@90: VampnProto::buildRpcRequest_Load(builder, request); c@90: ReqId id = getId(); c@90: builder.getId().setNumber(id); c@90: c@90: auto arr = messageToFlatArray(message); c@90: c@90: auto responseBuffer = m_transport->call(arr.asChars().begin(), c@90: arr.asChars().size()); c@90: c@90: //!!! ... --> will also need some way to kill this process c@90: //!!! (from another thread) c@90: c@90: auto karr = toKJArray(responseBuffer); c@90: capnp::FlatArrayMessageReader responseMessage(karr); c@90: RpcResponse::Reader reader = responseMessage.getRoot(); c@90: c@90: //!!! handle (explicit) error case c@90: c@90: checkResponseType(reader, RpcResponse::Response::Which::LOAD, id); c@90: c@90: const LoadResponse::Reader &lr = reader.getResponse().getLoad(); c@90: c@90: Vamp::HostExt::PluginStaticData psd; c@90: Vamp::HostExt::PluginConfiguration defaultConfig; c@90: VampnProto::readExtractorStaticData(psd, lr.getStaticData()); c@90: VampnProto::readConfiguration(defaultConfig, lr.getDefaultConfiguration()); c@90: c@90: Vamp::Plugin *plugin = new PiperStubPlugin(this, c@90: inputSampleRate, c@90: psd, c@90: defaultConfig); c@90: c@90: m_mapper.addPlugin(lr.getHandle(), plugin); c@90: c@90: return plugin; c@90: }; c@90: c@90: protected: c@90: virtual c@90: Vamp::Plugin::OutputList c@90: configure(PiperStubPlugin *plugin, c@90: Vamp::HostExt::PluginConfiguration config) override { c@90: c@90: if (!m_transport->isOK()) { c@90: throw std::runtime_error("Piper server failed to start"); c@90: } c@90: c@90: Vamp::HostExt::ConfigurationRequest request; c@90: request.plugin = plugin; c@90: request.configuration = config; c@90: c@90: capnp::MallocMessageBuilder message; c@90: RpcRequest::Builder builder = message.initRoot(); c@90: c@90: VampnProto::buildRpcRequest_Configure(builder, request, m_mapper); c@90: ReqId id = getId(); c@90: builder.getId().setNumber(id); c@90: c@90: auto arr = messageToFlatArray(message); c@90: auto responseBuffer = m_transport->call(arr.asChars().begin(), c@90: arr.asChars().size()); c@90: auto karr = toKJArray(responseBuffer); c@90: capnp::FlatArrayMessageReader responseMessage(karr); c@90: RpcResponse::Reader reader = responseMessage.getRoot(); c@90: c@90: //!!! handle (explicit) error case c@90: c@90: checkResponseType(reader, RpcResponse::Response::Which::CONFIGURE, id); c@90: c@90: Vamp::HostExt::ConfigurationResponse cr; c@90: VampnProto::readConfigurationResponse(cr, c@90: reader.getResponse().getConfigure(), c@90: m_mapper); c@90: c@90: return cr.outputs; c@90: }; c@90: c@90: virtual c@90: Vamp::Plugin::FeatureSet c@90: process(PiperStubPlugin *plugin, c@90: std::vector > inputBuffers, c@90: Vamp::RealTime timestamp) override { c@90: c@90: if (!m_transport->isOK()) { c@90: throw std::runtime_error("Piper server failed to start"); c@90: } c@90: c@90: Vamp::HostExt::ProcessRequest request; c@90: request.plugin = plugin; c@90: request.inputBuffers = inputBuffers; c@90: request.timestamp = timestamp; c@90: c@90: capnp::MallocMessageBuilder message; c@90: RpcRequest::Builder builder = message.initRoot(); c@90: c@90: VampnProto::buildRpcRequest_Process(builder, request, m_mapper); c@90: ReqId id = getId(); c@90: builder.getId().setNumber(id); c@90: c@90: auto arr = messageToFlatArray(message); c@90: auto responseBuffer = m_transport->call(arr.asChars().begin(), c@90: arr.asChars().size()); c@90: auto karr = toKJArray(responseBuffer); c@90: capnp::FlatArrayMessageReader responseMessage(karr); c@90: RpcResponse::Reader reader = responseMessage.getRoot(); c@90: c@90: //!!! handle (explicit) error case c@90: c@90: checkResponseType(reader, RpcResponse::Response::Which::PROCESS, id); c@90: c@90: Vamp::HostExt::ProcessResponse pr; c@90: VampnProto::readProcessResponse(pr, c@90: reader.getResponse().getProcess(), c@90: m_mapper); c@90: c@90: return pr.features; c@90: } c@90: c@90: virtual Vamp::Plugin::FeatureSet c@90: finish(PiperStubPlugin *plugin) override { c@90: c@90: if (!m_transport->isOK()) { c@90: throw std::runtime_error("Piper server failed to start"); c@90: } c@90: c@90: Vamp::HostExt::FinishRequest request; c@90: request.plugin = plugin; c@90: c@90: capnp::MallocMessageBuilder message; c@90: RpcRequest::Builder builder = message.initRoot(); c@90: c@90: VampnProto::buildRpcRequest_Finish(builder, request, m_mapper); c@90: ReqId id = getId(); c@90: builder.getId().setNumber(id); c@90: c@90: auto arr = messageToFlatArray(message); c@90: auto responseBuffer = m_transport->call(arr.asChars().begin(), c@90: arr.asChars().size()); c@90: auto karr = toKJArray(responseBuffer); c@90: capnp::FlatArrayMessageReader responseMessage(karr); c@90: RpcResponse::Reader reader = responseMessage.getRoot(); c@90: c@90: //!!! handle (explicit) error case c@90: c@90: checkResponseType(reader, RpcResponse::Response::Which::FINISH, id); c@90: c@90: Vamp::HostExt::ProcessResponse pr; c@90: VampnProto::readFinishResponse(pr, c@90: reader.getResponse().getFinish(), c@90: m_mapper); c@90: c@90: m_mapper.removePlugin(m_mapper.pluginToHandle(plugin)); c@90: c@90: // Don't delete the plugin. It's the plugin that is supposed c@90: // to be calling us here c@90: c@90: return pr.features; c@90: } c@90: c@90: private: c@90: AssignedPluginHandleMapper m_mapper; c@90: ReqId getId() { c@90: //!!! todo: mutex c@90: static ReqId m_nextId = 0; c@90: return m_nextId++; c@90: } c@90: c@90: kj::Array c@90: toKJArray(const std::vector &buffer) { c@90: // We could do this whole thing with fewer copies, but let's c@90: // see whether it matters first c@90: size_t wordSize = sizeof(capnp::word); c@90: size_t words = buffer.size() / wordSize; c@90: kj::Array karr(kj::heapArray(words)); c@90: memcpy(karr.begin(), buffer.data(), words * wordSize); c@90: return karr; c@90: } c@90: c@90: void c@90: checkResponseType(const RpcResponse::Reader &r, c@90: RpcResponse::Response::Which type, c@90: ReqId id) { c@90: c@90: if (r.getResponse().which() != type) { c@90: throw std::runtime_error("Wrong response type"); c@90: } c@90: if (ReqId(r.getId().getNumber()) != id) { c@90: throw std::runtime_error("Wrong response id"); c@90: } c@90: } c@90: c@90: private: c@90: SynchronousTransport *m_transport; //!!! I don't own this, but should I? c@90: }; c@90: c@90: } c@90: c@90: #endif