src/main/java/org/yaz4j/async/AsyncConnection.java

   1 /*
   2  * Copyright (c) 1995-2015, Index Data
   3  * All rights reserved.
   4  * See the file LICENSE for details.
   5  */
   6 package org.yaz4j.async;
   7
   8 import org.yaz4j.Connection;
   9 import org.yaz4j.Query;
  10 import org.yaz4j.Record;
  11 import org.yaz4j.ResultSet;
  12 import org.yaz4j.exception.ZoomException;
  13 import org.yaz4j.jni.SWIGTYPE_p_ZOOM_connection_p;
  14 import static org.yaz4j.jni.yaz4jlib.*;
  15 import org.yaz4j.util.Unstable;
  16
  17 /**
  18  * Represents an asynchronous connection, all methods of this class
  19  * (e.g connect, search) are non-blocking.
  20  *
  21  * Note that async support is missing for scan and extended services at this point.
  22  *
  23  * @author jakub
  24  */
  25 @Unstable
  26 public class AsyncConnection extends Connection {
  27   private ResultSet lastResultSet;
  28   ErrorHandler eh;
  29   //make sure error is only handled once
  30   boolean errorHandled = false;
  31   int handledRecordOffset = 0;
  32   ErrorHandler reh;
  33   SearchHandler sh;
  34   RecordHandler rh;
  35
  36   /**
  37    * Invoked immediately in response to the search request.
  38    *
  39    * Allows to read things the hit count and facets. Records should be read
  40    * and processed in the record handler.
  41    */
  42   public interface SearchHandler {
  43     public void handle(ResultSet rs);
  44   }
  45
  46   /**
  47    * Invoked for every retrieved record.
  48    */
  49   public interface RecordHandler {
  50     public void handle(Record r);
  51   }
  52
  53   /**
  54    * Invoked for any protocol or connection level error.
  55    */
  56   public interface ErrorHandler {
  57     public void handle(ZoomException e);
  58   }
  59
  60   public AsyncConnection(String host, int port) {
  61     super(host, port);
  62     ZOOM_connection_option_set(zoomConnection, "async", "1");
  63     closed = false;
  64   }
  65
  66   @Override
  67   public ResultSet search(Query query) throws ZoomException {
  68     errorHandled = false;
  69     lastResultSet = super.search(query);
  70     return null;
  71   }
  72
  73   /**
  74    * Register a hander for the search response.
  75    * @param sh search response handler
  76    * @return
  77    */
  78   public AsyncConnection onSearch(SearchHandler sh) {
  79     this.sh = sh;
  80     return this;
  81   }
  82
  83   /**
  84    * Register a handler for each retrieved record.
  85    * @param rh record handler
  86    * @return
  87    */
  88   public AsyncConnection onRecord(RecordHandler rh) {
  89     this.rh = rh;
  90     return this;
  91   }
  92
  93   /**
  94    * Register a handler for a connection level errors.
  95    * @param eh error handler
  96    * @return
  97    */
  98   public AsyncConnection onError(ErrorHandler eh) {
  99     this.eh = eh;
 100     return this;
 101   }
 102
 103   /**
 104    * Register a handler for record level errors (e.g decoding).
 105    * @param reh record error handler
 106    * @return
 107    */
 108   public AsyncConnection onRecordError(ErrorHandler reh) {
 109     this.reh = reh;
 110     return this;
 111   }
 112
 113   //actuall handler, pkg-private
 114
 115   void handleSearch() {
 116     handleError();
 117     //handle search
 118     if (sh != null) sh.handle(lastResultSet);
 119   }
 120
 121   void handleRecord() {
 122     //TODO clone the record to detach it from the result set
 123     try {
 124       if (rh != null) rh.handle(lastResultSet.getRecord(handledRecordOffset));
 125     } catch (ZoomException ex) {
 126       if (reh != null) reh.handle(ex);
 127     } finally {
 128       handledRecordOffset++;
 129     }
 130   }
 131
 132   void handleError() {
 133     //handle error
 134     if (!errorHandled) {
 135       ZoomException err = getZoomException();
 136       if (err != null) {
 137         if (eh != null) {
 138           eh.handle(err);
 139           errorHandled = true;
 140         }
 141       }
 142     }
 143   }
 144
 145   /**
 146    * Expose native connection to the async package, keep it package private.
 147    */
 148   SWIGTYPE_p_ZOOM_connection_p getNativeConnection() {
 149     return zoomConnection;
 150   }
 151
 152 }