Changeset 2122 for trunk/sources/HeuristicLab.Hive.Server.Core/3.2/ClientCommunicator.cs

Timestamp:

06/28/09 14:28:30 (15 years ago)

Author:

svonolfe

Message:

Fixed issue related to the streaming of results (#680)

File:

• trunk/sources/HeuristicLab.Hive.Server.Core/3.2/ClientCommunicator.cs (modified) (13 diffs)

trunk/sources/HeuristicLab.Hive.Server.Core/3.2/ClientCommunicator.cs

@@ -37 +37 @@
 using HeuristicLab.DataAccess.Interfaces;
 using System.IO;
+using System.Runtime.Serialization.Formatters.Binary;
 
 namespace HeuristicLab.Hive.Server.Core {
@@ -465 +466 @@
 
     public ResponseResultReceived ProcessJobResult(
-      JobResult result,
       Stream stream,
       bool finished) {
@@ -474 +474 @@
 
       try {
+        BinaryFormatter formatter =
+          new BinaryFormatter();
+
+        JobResult result =
+          (JobResult)formatter.Deserialize(stream);
+
         tx = session.BeginTransaction();
 
@@ -485 +491 @@
           finished);
 
-        //second deserialize the BLOB
-        IJobResultsAdapter jobResultsAdapter =
-          session.GetDataAdapter<JobResult, IJobResultsAdapter>();
-
-        IJobAdapter jobAdapter =
-          session.GetDataAdapter<Job, IJobAdapter>();
-
-        jobResultStream =
-          jobResultsAdapter.GetSerializedJobResultStream(result.Id, true);
-
-        jobStream =
-          jobAdapter.GetSerializedJobStream(result.JobId, true);
-
-        byte[] buffer = new byte[3024];
-        int read = 0;
-        while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {
-          jobResultStream.Write(buffer, 0, read);
-
-          if (finished)
+        if (response.Success) {
+          //second deserialize the BLOB
+          IJobResultsAdapter jobResultsAdapter =
+            session.GetDataAdapter<JobResult, IJobResultsAdapter>();
+
+          IJobAdapter jobAdapter =
+            session.GetDataAdapter<Job, IJobAdapter>();
+
+          jobResultStream =
+            jobResultsAdapter.GetSerializedJobResultStream(response.JobResultId, true);
+
+          jobStream =
+            jobAdapter.GetSerializedJobStream(result.JobId, true);
+
+          byte[] buffer = new byte[3024];
+          int read = 0;
+          while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {
+            jobResultStream.Write(buffer, 0, read);
             jobStream.Write(buffer, 0, read);
-        }
-
-        jobStream.Close();
-
-        tx.Commit();
+          }
+
+          jobResultStream.Close();
+          jobStream.Close();
+
+          tx.Commit();
+        }
 
         return response;
@@ -565 +572 @@
           response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOB_WITH_THIS_ID;
           response.JobId = jobId;
+          tx.Rollback();
           return response;
         }
@@ -570 +578 @@
           response.Success = false;
           response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_WAS_ABORTED;
+          tx.Rollback();
+          return response;
         }
         if (job.JobInfo.Client == null) {
@@ -575 +585 @@
           response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
           response.JobId = jobId;
+          tx.Rollback();
           return response;
         }
@@ -581 +592 @@
           response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_CLIENT_FOR_JOB;
           response.JobId = jobId;
+          tx.Rollback();
           return response;
         }
@@ -587 +599 @@
           response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
           response.JobId = jobId;
+          tx.Rollback();
           return response;
         }
@@ -597 +610 @@
           response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
           response.JobId = jobId;
+          tx.Rollback();
           return response;
         }
@@ -602 +616 @@
 
         if (finished) {
-          job.JobInfo.State = State.finished;
-          job.SerializedJobData = result;
-          jobAdapter.UpdateSerializedJob(job);
-        }
+          job.JobInfo.State = State.finished; 
+        }
+
+        job.SerializedJobData = result;
+        jobAdapter.UpdateSerializedJob(job);
 
         List<JobResult> jobResults = new List<JobResult>(
@@ -625 +640 @@
 
         jobResultAdapter.UpdateSerializedJobResult(serializedjobResult);
-        jobAdapter.Update(job.JobInfo);
 
         response.Success = true;
@@ -631 +645 @@
         response.JobId = jobId;
         response.finished = finished;
+        response.JobResultId = jobResult.Id;
 
         tx.Commit();