Changeset 2117 for trunk/sources/HeuristicLab.Hive.Client.Communication

Timestamp:

06/26/09 12:18:32 (16 years ago)

Author:

svonolfe

Message:

Streaming of Jobs and JobsResults directly from/to the DB (#680)

Location:

trunk/sources/HeuristicLab.Hive.Client.Communication/3.2

Files:

• Service References/ServerService/Reference.cs (modified) (1 diff)
• WcfService.cs (modified) (5 diffs)

trunk/sources/HeuristicLab.Hive.Client.Communication/3.2/Service References/ServerService/Reference.cs

@@ -184 +184 @@
             }
         }
+
+        public byte[] Data {
+          get {
+            base.RaiseExceptionIfNecessary();
+            return ((byte[])(this.results[1]));
+          }
+        }
     }
     

trunk/sources/HeuristicLab.Hive.Client.Communication/3.2/WcfService.cs

@@ -194 +194 @@
     void proxy_SendStreamedJobCompleted(object sender, SendStreamedJobCompletedEventArgs e) {
       if (e.Error == null) {
-        Stream stream =
-          (Stream)e.Result;
-                
-        BinaryFormatter formatter =
-          new BinaryFormatter();
-        ResponseJob response = (ResponseJob)formatter.Deserialize(stream);
-
-        SendJobCompletedEventArgs completedEventArgs =
-          new SendJobCompletedEventArgs(new object[] { response }, e.Error, e.Cancelled, e.UserState);
-        SendJobCompleted(sender, completedEventArgs);
+        Stream stream = null;
+
+        try {
+          stream = (Stream)e.Result;
+
+          //first deserialize the response
+          BinaryFormatter formatter =
+            new BinaryFormatter();
+          ResponseJob response =
+            (ResponseJob)formatter.Deserialize(stream);
+
+          //second deserialize the BLOB
+          MemoryStream memStream = new MemoryStream();
+          byte[] buffer = new byte[3024];
+          int read = 0;
+          while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {
+            memStream.Write(buffer, 0, read);
+          }
+
+          SendJobCompletedEventArgs completedEventArgs =
+            new SendJobCompletedEventArgs(new object[] { response, memStream.GetBuffer() }, e.Error, e.Cancelled, e.UserState);
+          SendJobCompleted(sender, completedEventArgs);
+        }
+        finally {
+          if(stream != null)
+            stream.Dispose();
+        }
       } else
         HandleNetworkError(e.Error);
@@ -216 +233 @@
     public event System.EventHandler<StoreFinishedJobResultCompletedEventArgs> StoreFinishedJobResultCompleted;
     public void StoreFinishedJobResultAsync(Guid clientId, Guid jobId, byte[] result, double percentage, Exception exception, bool finished) {
-      if (ConnState == NetworkEnum.WcfConnState.Loggedin)
-        proxy.StoreFinishedJobResultStreamedAsync(
-          GetStreamedJobResult(clientId, jobId, result, percentage, exception));
-    }
+      if (ConnState == NetworkEnum.WcfConnState.Loggedin) {
+        Stream stream =
+          GetStreamedJobResult(clientId, jobId, result, percentage, exception);
+
+        proxy.StoreFinishedJobResultStreamedAsync(stream, stream);
+      }
+     }
     private void proxy_StoreFinishedJobResultStreamedCompleted(object sender, StoreFinishedJobResultStreamedCompletedEventArgs e) {
+      Stream stream =
+        (Stream)e.UserState;
+      if (stream != null)
+        stream.Dispose();
+      
       if (e.Error == null) {
         StoreFinishedJobResultCompletedEventArgs args =
@@ -235 +260 @@
     public event System.EventHandler<ProcessSnapshotCompletedEventArgs> ProcessSnapshotCompleted;
     public void ProcessSnapshotAsync(Guid clientId, Guid jobId, byte[] result, double percentage, Exception exception, bool finished) {
-      if(ConnState == NetworkEnum.WcfConnState.Loggedin)
-        proxy.ProcessSnapshotStreamedAsync(
-          GetStreamedJobResult(
-            clientId, jobId, result, percentage, exception));
+      if (ConnState == NetworkEnum.WcfConnState.Loggedin) {
+        Stream stream = GetStreamedJobResult(
+            clientId, jobId, result, percentage, exception);
+
+        proxy.ProcessSnapshotStreamedAsync(stream, stream);
+      }
     }
     void proxy_ProcessSnapshotStreamedCompleted(object sender, ProcessSnapshotStreamedCompletedEventArgs e) {
+      Stream stream =
+        (Stream)e.UserState;
+      if (stream != null)
+        stream.Dispose();
+      
       if (e.Error == null) {
         ProcessSnapshotCompletedEventArgs args =
@@ -283 +315 @@
     /// </summary>
     private Stream GetStreamedJobResult(Guid clientId, Guid jobId, byte[] result, double percentage, Exception exception) {
-      SerializedJobResult serializedJobResult =
-          new SerializedJobResult();
-      JobResult jobResult = new JobResult();
+      JobResult jobResult =
+          new JobResult();
       jobResult.ClientId = clientId;
       jobResult.JobId = jobId;
@@ -291 +322 @@
       jobResult.Exception = exception;
 
-      serializedJobResult.JobResult = jobResult;
-      serializedJobResult.SerializedJobResultData = result;
-
-      MemoryStream stream =
-        new MemoryStream();
-
-      BinaryFormatter formatter =
-        new BinaryFormatter();
-
-      formatter.Serialize(stream, serializedJobResult);
-      stream.Seek(0, SeekOrigin.Begin);
+      MultiStream stream =
+              new MultiStream();
+
+      //first send result
+      stream.AddStream(
+        new StreamedObject<JobResult>(jobResult));
+
+      //second stream the job binary data
+      MemoryStream memStream =
+        new MemoryStream(result, false);
+      stream.AddStream(memStream);
 
       return stream;