rgw: organize get_obj handling for copy a bit different

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

5 files changed, 71 insertions, 16 deletions

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 317a1b1a029..14ebed14c3b 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -1886,6 +1886,17 @@ bool RGWRados::aio_completed(void *handle)
   AioCompletion *c = (AioCompletion *)handle;
   return c->is_complete();
 }
+
+class RGWRadosPutObj : public RGWGetDataCB
+{
+  rgw_obj obj;
+public:
+  RGWRadosPutObj(rgw_obj& _o) : obj(_o) {}
+  int handle_data(bufferlist& bl, off_t ofs, off_t len) {
+    return 0;
+  }
+};
+
 /**
  * Copy an object.
  * dest_obj: the object to copy into
@@ -1952,15 +1963,11 @@ int RGWRados::copy_obj(void *ctx,
     map<string, bufferlist> src_attrs;
 
     RGWRESTStreamReadRequest *in_stream_req;
+    RGWRadosPutObj cb(dest_obj);
   
-    int ret = rest_conn->get_obj_init(user_id, src_obj, &in_stream_req);
-    if (ret < 0)
-      return ret;
-#if 0
-    ret = get_obj_iterate(ctx, &handle, src_obj, 0, astate->size - 1, out_stream_req->get_out_cb());
+    int ret = rest_conn->get_obj(user_id, src_obj, &cb, &in_stream_req);
     if (ret < 0)
       return ret;
-#endif
 
     string etag;
 
diff --git a/src/rgw/rgw_rest_client.cc b/src/rgw/rgw_rest_client.cc
index 863b7147c41..9ff6ba57d8e 100644
--- a/src/rgw/rgw_rest_client.cc
+++ b/src/rgw/rgw_rest_client.cc
@@ -529,7 +529,7 @@ int RGWRESTStreamWriteRequest::complete(string& etag, time_t *mtime)
   return status;
 }
 
-int RGWRESTStreamReadRequest::get_obj_init(RGWAccessKey& key, rgw_obj& obj)
+int RGWRESTStreamReadRequest::get_obj(RGWAccessKey& key, rgw_obj& obj)
 {
   string resource = obj.bucket.name + "/" + obj.object;
   string new_url = url;
@@ -596,7 +596,51 @@ int RGWRESTStreamReadRequest::complete(string& etag, time_t *mtime)
   return status;
 }
 
-int RGWRESTStreamReadRequest::send_data(void *ptr, size_t len) {
+int RGWRESTStreamReadRequest::receive_data(void *ptr, size_t len)
+{
+  bufferptr bp((const char *)ptr, len);
+  bufferlist bl;
+  bl.append(bp);
+  int ret = cb->handle_data(bl, ofs, len);
+  if (ret < 0)
+    return ret;
+  ofs += len;
+  return len;
+#if 0
+  return cb->handle_data(bl
+  const char *p = (const char *)ptr;
+  size_t orig_len = len;
+  while (len > 0) {
+    size_t read_len = RGW_MAX_CHUNK_SIZE - chunk_ofs;
+    if (read_len > len)
+      read_len = len;
+
+    bufferptr bp((const char *)p, read_len);
+    in_data.append(bp);
+
+    p += read_len;
+    len -= read_len;
+    chunk_ofs += read_len;
+    if (chunk_ofs == RGW_MAX_CHUNK_SIZE) {
+      chunk_ofs = 0;
+      size_t data_len = in_data.length();
+      int r = cb->handle_data(in_data, ofs, data_len);
+      if (r < 0)
+        return r;
+
+      ofs += data_len;
+
+      in_data.clear();
+    }
+  }
+
+  return orig_len;
+#endif
+}
+
+int RGWRESTStreamReadRequest::send_data(void *ptr, size_t len)
+{
+  /* not sending any data */
   return 0;
 }
 
diff --git a/src/rgw/rgw_rest_client.h b/src/rgw/rgw_rest_client.h
index a2c14b42cb5..973a88e7972 100644
--- a/src/rgw/rgw_rest_client.h
+++ b/src/rgw/rgw_rest_client.h
@@ -74,16 +74,20 @@ public:
 
 class RGWRESTStreamReadRequest : public RGWRESTSimpleRequest {
   Mutex lock;
-  void *handle;
   RGWGetDataCB *cb;
+  bufferlist in_data;
+  size_t chunk_ofs;
+  size_t ofs;
 public:
   int send_data(void *ptr, size_t len);
+  int receive_data(void *ptr, size_t len);
 
-  RGWRESTStreamReadRequest(CephContext *_cct, string& _url, list<pair<string, string> > *_headers,
+  RGWRESTStreamReadRequest(CephContext *_cct, string& _url, RGWGetDataCB *_cb, list<pair<string, string> > *_headers,
                 list<pair<string, string> > *_params) : RGWRESTSimpleRequest(_cct, _url, _headers, _params),
-                lock("RGWRESTStreamReadRequest"), handle(NULL), cb(NULL) {}
+                lock("RGWRESTStreamReadRequest"), cb(_cb),
+                chunk_ofs(0), ofs(0) {}
   ~RGWRESTStreamReadRequest() {}
-  int get_obj_init(RGWAccessKey& key, rgw_obj& obj);
+  int get_obj(RGWAccessKey& key, rgw_obj& obj);
   int complete(string& etag, time_t *mtime);
 
   void set_in_cb(RGWGetDataCB *_cb) { cb = _cb; }
diff --git a/src/rgw/rgw_rest_conn.cc b/src/rgw/rgw_rest_conn.cc
index fc6f23b0350..fb3de5b423d 100644
--- a/src/rgw/rgw_rest_conn.cc
+++ b/src/rgw/rgw_rest_conn.cc
@@ -69,7 +69,7 @@ int RGWRegionConnection::complete_request(RGWRESTStreamWriteRequest *req, string
   return ret;
 }
 
-int RGWRegionConnection::get_obj_init(const string& uid, rgw_obj& obj, RGWRESTStreamReadRequest **req)
+int RGWRegionConnection::get_obj(const string& uid, rgw_obj& obj, RGWGetDataCB *cb, RGWRESTStreamReadRequest **req)
 {
   string url;
   int ret = get_url(url);
@@ -79,8 +79,8 @@ int RGWRegionConnection::get_obj_init(const string& uid, rgw_obj& obj, RGWRESTSt
   list<pair<string, string> > params;
   params.push_back(make_pair<string, string>(RGW_SYS_PARAM_PREFIX "uid", uid));
   params.push_back(make_pair<string, string>(RGW_SYS_PARAM_PREFIX "region", region));
-  *req = new RGWRESTStreamReadRequest(cct, url, NULL, &params);
-  return (*req)->get_obj_init(key, obj);
+  *req = new RGWRESTStreamReadRequest(cct, url, cb, NULL, &params);
+  return (*req)->get_obj(key, obj);
 }
 
 int RGWRegionConnection::complete_request(RGWRESTStreamReadRequest *req, string& etag, time_t *mtime)
diff --git a/src/rgw/rgw_rest_conn.h b/src/rgw/rgw_rest_conn.h
index f9527afe8a1..f0ef6ed0247 100644
--- a/src/rgw/rgw_rest_conn.h
+++ b/src/rgw/rgw_rest_conn.h
@@ -28,7 +28,7 @@ public:
                    map<string, bufferlist>& attrs, RGWRESTStreamWriteRequest **req);
   int complete_request(RGWRESTStreamWriteRequest *req, string& etag, time_t *mtime);
 
-  int get_obj_init(const string& uid, rgw_obj& obj, RGWRESTStreamReadRequest **req);
+  int get_obj(const string& uid, rgw_obj& obj, RGWGetDataCB *cb, RGWRESTStreamReadRequest **req);
   int complete_request(RGWRESTStreamReadRequest *req, string& etag, time_t *mtime);
 };