class ClientDataStreamerAddDataRequest extends ClientDataStreamerRequest {
    /** */
    private final long streamerId;

    /** */
    private final byte flags;

    /** */
    private final Collection<DataStreamerEntry> entries;

    /**
     * Constructor.
     *
     * @param reader Data reader.
     */
    public ClientDataStreamerAddDataRequest(BinaryReaderExImpl reader) {
        super(reader);

        streamerId = reader.readLong();
        flags = reader.readByte();
        entries = ClientDataStreamerReader.read(reader);
    }

    /**
     * {@inheritDoc}
     */
    @Override public ClientResponse process(ClientConnectionContext ctx) {<FILL_FUNCTION_BODY>}
}

        ClientDataStreamerHandle handle = ctx.resources().get(streamerId);
        DataStreamerImpl<KeyCacheObject, CacheObject> dataStreamer =
                (DataStreamerImpl<KeyCacheObject, CacheObject>)handle.getStreamer();

        try {
            if (entries != null)
                dataStreamer.addData(entries);

            if ((flags & FLUSH) != 0)
                dataStreamer.flush();

            if ((flags & CLOSE) != 0) {
                dataStreamer.close();
                ctx.resources().release(streamerId);
            }
        }
        catch (IllegalStateException unused) {
            return getInvalidNodeStateResponse();
        }

        return new ClientResponse(requestId());
    
<no_super_class>