feat(experimental): flush the last chunk in append method (#1699)

Earlier the last chunk was being flushed while calling the close()
method. Now it will be done inside the append method itself.

Author: Pulkit0110

google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py

@@ -188,8 +188,9 @@ async def append(self, data: bytes) -> None:
         ie. `self.offset` bytes relative to the begining of the object.
 
         This method sends the provided `data` to the GCS server in chunks.
-        and persists data in GCS at every `_MAX_BUFFER_SIZE_BYTES` bytes by
-        calling `self.simple_flush`.
+        and persists data in GCS at every `_DEFAULT_FLUSH_INTERVAL_BYTES` bytes
+        or at the last chunk whichever is earlier. Persisting is done by setting
+        `flush=True` on request.
 
         :type data: bytes
         :param data: The bytes to append to the object.
@@ -214,20 +215,33 @@ async def append(self, data: bytes) -> None:
         while start_idx < total_bytes:
             end_idx = min(start_idx + _MAX_CHUNK_SIZE_BYTES, total_bytes)
             data_chunk = data[start_idx:end_idx]
+            is_last_chunk = end_idx == total_bytes
+            chunk_size = end_idx - start_idx
             await self.write_obj_stream.send(
                 _storage_v2.BidiWriteObjectRequest(
                     write_offset=self.offset,
                     checksummed_data=_storage_v2.ChecksummedData(
                         content=data_chunk,
                         crc32c=int.from_bytes(Checksum(data_chunk).digest(), "big"),
                     ),
+                    state_lookup=is_last_chunk,
+                    flush=is_last_chunk
+                    or (
+                        self.bytes_appended_since_last_flush + chunk_size
+                        >= self.flush_interval
+                    ),
                 )
             )
-            chunk_size = end_idx - start_idx
             self.offset += chunk_size
             self.bytes_appended_since_last_flush += chunk_size
+
             if self.bytes_appended_since_last_flush >= self.flush_interval:
-                await self.simple_flush()
+                self.bytes_appended_since_last_flush = 0
+
+            if is_last_chunk:
+                response = await self.write_obj_stream.recv()
+                self.persisted_size = response.persisted_size
+                self.offset = self.persisted_size
                 self.bytes_appended_since_last_flush = 0
             start_idx = end_idx
 
@@ -292,20 +306,24 @@ async def close(self, finalize_on_close=False) -> Union[int, _storage_v2.Object]
             raise ValueError("Stream is not open. Call open() before close().")
 
         if finalize_on_close:
-            await self.finalize()
-        else:
-            await self.flush()
+            return await self.finalize()
 
         await self.write_obj_stream.close()
 
         self._is_stream_open = False
         self.offset = None
-        return self.object_resource if finalize_on_close else self.persisted_size
+        return self.persisted_size
 
     async def finalize(self) -> _storage_v2.Object:
         """Finalizes the Appendable Object.
 
         Note: Once finalized no more data can be appended.
+        This method is different from `close`. if `.close()` is called data may
+        still be appended to object at a later point in time by opening with
+        generation number.
+        (i.e. `open(..., generation=<object_generation_number>)`.
+        However if `.finalize()` is called no more data can be appended to the
+        object.
 
         rtype: google.cloud.storage_v2.types.Object
         returns: The finalized object resource.
@@ -322,6 +340,10 @@ async def finalize(self) -> _storage_v2.Object:
         response = await self.write_obj_stream.recv()
         self.object_resource = response.resource
         self.persisted_size = self.object_resource.size
+        await self.write_obj_stream.close()
+
+        self._is_stream_open = False
+        self.offset = None
         return self.object_resource
 
     # helper methods.