Retrofit2 и потоковые сериализованные объекты

Question

Я использую Retrofit2. До сих пор я пытался отправить и получить java.util.List<MyCustomClass>, и он работает очень хорошо. Мне, однако, интересно, что будет с действительно большими списками. Есть ли способ сделать более «потоковый» интерфейс?

Например, я мог бы проанализировать api Iterator<MyCustomClass> вместо List и создать каждый следующий экземпляр на лету для сохранения памяти при отправке данных в API. То же самое Iterator<MyCustomClass> использовать в противоположном направлении.

Есть ли способ достичь этого?

Я вижу, что Gson способен выполнять сериализацию потокового вещания и десериализацию. Можно ли использовать эту технологию?

Редактировать: Уточненный вопрос.

Answer 1

Интересный вопрос. Прежде всего, вы должны учитывать, что List и Iterable отличаются семантически, особенно в области ввода-вывода. Если вы посмотрите на него с точки зрения ввода-вывода, списки используются для сбора элементов в объект в памяти, а затем немедленно закрывают базовый ресурс. Я думаю, что итераторы не знакомы с Gson из-за того, что они слишком ленивы и откладывают оценку элементов от входных потоков: итераторы должны иметь возможность возвращать новые итераторы по дизайну. Невозможно создать новый итератор, если не запрашивается другое чтение ресурсов ввода-вывода. Здесь возникает путаница. Недавно я решил очень похожую проблему для Spring Framework (MVC-модуля) и потоков Java 8, и у меня, похоже, есть решение для Retrofit. Однако Retrofit необходимо использовать очень осторожно, чтобы правильно использовать его и защищать от утечек ресурсов. Обратите внимание: здесь вы не можете использовать итераторы, но вы можете использовать итераторы (Iterator): они семантически очень похожи на потоки Java 8 (Stream) и потоки ввода-вывода Java (InputStream), так как они не могут be повторно. Однако, зная разницу, вы все еще можете работать с экземплярами Iterator.

Давайте создадим простой статический HTTP веб-сервер в Python, если вы хотите поэкспериментировать:

from BaseHTTPServer import BaseHTTPRequestHandler,HTTPServer 

PORT_NUMBER = 8080 

class myHandler(BaseHTTPRequestHandler): 

    def do_GET(self): 
     self.send_response(200) 
     self.send_header('Content-type','text/json') 
     self.end_headers() 
     # Send the html message 
     self.wfile.write('["foo","bar","baz"]') 
     return 

try: 
    server = HTTPServer(('', PORT_NUMBER), myHandler) 
    print 'Started httpserver on port ' , PORT_NUMBER 
    server.serve_forever() 
except KeyboardInterrupt: 
    print '^C received, shutting down the web server' 
    server.socket.close() 

Это веб-сервис всегда возвращает ["foo","bar","baz"] не принимая во внимание запросы. Теперь создать службу образца для работы с ним:

interface IService { 

    @GET("/") 
    Call<Iterator<String>> get(); 

} 

Во-вторых, Gson не работает с итерируемыми или итераторы по умолчанию, так что пользовательский (де) модуль сериализации должен быть реализован. Тип адаптера завод ниже либо создает специальный адаптер типа для итераторов, или позволяет Gson выбрать другую стратегию ранее, если таковая существует:

final class IteratorTypeAdapterFactory 
     implements TypeAdapterFactory { 

    private static final TypeAdapterFactory iteratorTypeAdapterFactory = new IteratorTypeAdapterFactory(); 

    private IteratorTypeAdapterFactory() { 
    } 

    static TypeAdapterFactory getIteratorTypeAdapterFactory() { 
     return iteratorTypeAdapterFactory; 
    } 

    @Override 
    public <T> TypeAdapter<T> create(final Gson gson, final TypeToken<T> type) { 
     if (Iterator.class.isAssignableFrom(type.getRawType())) { 
      @SuppressWarnings("unchecked") 
      final TypeAdapter<T> castTypeAdapter = (TypeAdapter<T>) getIteratorTypeAdapter(getIteratorParameterType(type.getType()), gson); 
      return castTypeAdapter; 
     } 
     return null; 
    } 

} 

Далее, реализация адаптера типа итератора заключается в следующем. Обратите внимание, что это очень легко писать итератор в поток, но чтение гораздо сложнее:

final class IteratorTypeAdapter<T> 
     extends TypeAdapter<Iterator<T>> { 

    private final Type elementType; 
    private final Gson gson; 

    private IteratorTypeAdapter(final Type elementType, final Gson gson) { 
     this.elementType = elementType; 
     this.gson = gson; 
    } 

    static <T> IteratorTypeAdapter<Iterator<T>> getIteratorTypeAdapter(final Type elementType, final Gson gson) { 
     return new IteratorTypeAdapter<>(elementType, gson); 
    } 

    @Override 
    @SuppressWarnings("resource") 
    public void write(final JsonWriter out, final Iterator<T> iterator) 
      throws IOException { 
     out.beginArray(); 
     while (iterator.hasNext()) { 
      final T next = iterator.next(); 
      gson.toJson(next, elementType, out); 
     } 
     out.endArray(); 
    } 

    @Override 
    public Iterator<T> read(final JsonReader in) { 
     return getJsonReaderIterator(elementType, gson, in); 
    } 

} 

И чтение итератора:

final class JsonReaderIterator<T> 
     implements Iterator<T>, Closeable { 

    private final Type elementType; 
    private final Gson gson; 
    private final JsonReader in; 

    private ReadingIteratorState state = ReadingIteratorState.BEFORE_ARRAY; 

    private JsonReaderIterator(final Type elementType, final Gson gson, final JsonReader in) { 
     this.elementType = elementType; 
     this.gson = gson; 
     this.in = in; 
    } 

    static <T> Iterator<T> getJsonReaderIterator(final Type elementType, final Gson gson, final JsonReader in) { 
     return new JsonReaderIterator<>(elementType, gson, in); 
    } 

    @Override 
    public boolean hasNext() { 
     try { 
      if (state == ReadingIteratorState.END_OF_STREAM) { 
       return false; 
      } 
      final boolean hasNext; 
      loop: 
      for (; ;) { 
       switch (state) { 
       case BEFORE_ARRAY: 
        if (in.peek() == BEGIN_ARRAY) { 
         in.beginArray(); 
         state = ReadingIteratorState.WITHIN_ARRAY; 
        } 
        continue; 
       case WITHIN_ARRAY: 
        if (in.peek() == END_ARRAY) { 
         in.endArray(); 
         state = ReadingIteratorState.END_OF_STREAM; 
         continue; 
        } 
        hasNext = true; 
        break loop; 
       case AFTER_ARRAY: 
        hasNext = false; 
        state = ReadingIteratorState.END_OF_STREAM; 
        break loop; 
       case END_OF_STREAM: 
        hasNext = false; 
        break loop; 
       default: 
        throw new AssertionError(state); 
       } 
      } 
      return hasNext; 
     } catch (final IOException ex) { 
      throw new RuntimeException(ex); 
     } 
    } 

    @Override 
    public T next() { 
     try { 
      if (!in.hasNext() || state == ReadingIteratorState.END_OF_STREAM) { 
       throw new NoSuchElementException(); 
      } 
      final T element; 
      loop: 
      for (; ;) { 
       switch (state) { 
       case BEFORE_ARRAY: 
        in.beginArray(); 
        state = ReadingIteratorState.WITHIN_ARRAY; 
        if (in.peek() == END_ARRAY) { 
         state = ReadingIteratorState.END_OF_STREAM; 
        } 
        break; 
       case WITHIN_ARRAY: 
        element = gson.fromJson(in, elementType); 
        if (in.peek() == END_ARRAY) { 
         state = ReadingIteratorState.AFTER_ARRAY; 
        } 
        break loop; 
       case AFTER_ARRAY: 
        in.endArray(); 
        state = ReadingIteratorState.END_OF_STREAM; 
        break; 
       case END_OF_STREAM: 
        throw new NoSuchElementException(String.valueOf(state)); 
       default: 
        throw new AssertionError(state); 
       } 
      } 
      return element; 
     } catch (final IOException ex) { 
      throw new RuntimeException(ex); 
     } 
    } 

    @Override 
    public void close() 
      throws IOException { 
     in.close(); 
    } 

    private enum ReadingIteratorState { 

     BEFORE_ARRAY, 
     WITHIN_ARRAY, 
     AFTER_ARRAY, 
     END_OF_STREAM 

    } 

} 

Обратите внимание, что итератор должен быть закрываемой для того, чтобы закрыть основной Ресурсы.Честно говоря, мне не нравится контракт метода close на Java, который требует, чтобы метод закрывал основные ресурсы, и я считаю, что закрытие ресурсов является обязанностью объекта, который открыл ресурсы. Например, такой итератор не нужно закрывать в Spring MVC, потому что фреймворк прослушивает сам HTTP-запрос, позволяет обработчикам обрабатывать запросы, а затем сам закрывает запросы. В Retrofit утечка ресурсов может произойти, если выставлять итераторы, поэтому именно поэтому эта реализация имеет реализацию метода close.

Следующее: настройка Модернизация для работы с Iterator -aware Gson. Ниже приведена фабрика преобразователей, которая может работать с ленивыми итераторами. Обратите внимание, что реализация по умолчанию GsonConverterFactory не позволяет работать с итераторами, поскольку она закрывает входные потоки до, и их можно обрабатывать и конвертировать в итераторы. Остальные два метода могут быть повторно использованы из реализации по умолчанию, но нет необходимости передавать его извне и могут быть созданы в частном порядке. Также обратите внимание, что преобразователи отклика фабрики работают только для самых лучших объектов, и нет необходимости закрывать итераторы, которые могут быть полями других объектов.

final class CustomGsonConverterFactory 
     extends Factory { 

    private final Gson gson; 
    private final GsonConverterFactory gsonConverterFactory; 

    private CustomGsonConverterFactory(final Gson gson, final GsonConverterFactory gsonConverterFactory) { 
     this.gson = gson; 
     this.gsonConverterFactory = gsonConverterFactory; 
    } 

    static Factory getCustomGsonConverterFactory(final Gson gson, final GsonConverterFactory gsonConverterFactory) { 
     return new CustomGsonConverterFactory(gson, gsonConverterFactory); 
    } 

    @Override 
    public Converter<ResponseBody, ?> responseBodyConverter(final Type type, final Annotation[] annotations, final Retrofit retrofit) { 
     final TypeAdapter<?> adapter = gson.getAdapter(TypeToken.get(type)); 
     final boolean isClosedElsewhere = isIterator(type); 
     return (Converter<ResponseBody, Object>) responseBody -> { 
      try { 
       return adapter.read(gson.newJsonReader(responseBody.charStream())); 
      } finally { 
       if (!isClosedElsewhere) { 
        responseBody.close(); 
       } 
      } 
     }; 
    } 

    @Override 
    public Converter<?, RequestBody> requestBodyConverter(final Type type, final Annotation[] parameterAnnotations, final Annotation[] methodAnnotations, 
      final Retrofit retrofit) { 
     return gsonConverterFactory.requestBodyConverter(type, parameterAnnotations, methodAnnotations, retrofit); 
    } 

    @Override 
    public Converter<?, String> stringConverter(final Type type, final Annotation[] annotations, final Retrofit retrofit) { 
     return gsonConverterFactory.stringConverter(type, annotations, retrofit); 
    } 

} 

Код выше использует некоторые отражения утилиты:

final class Reflection { 

    private Reflection() { 
    } 

    static Type getIteratorParameterType(final Type type) 
      throws IllegalArgumentException { 
     return getTParameterType(type, Iterator.class); 
    } 

    static boolean isIterator(final Type type) { 
     return Iterator.class.equals(type) 
       || type instanceof ParameterizedType && Iterator.class.equals(((ParameterizedType) type).getRawType()); 
    } 

    private static Type getTParameterType(final Type type, final Type expectedParameterizedType) 
      throws IllegalArgumentException { 
     if (expectedParameterizedType.equals(type)) { 
      return expectedParameterizedType; 
     } 
     if (type instanceof ParameterizedType) { 
      final ParameterizedType parameterizedType = (ParameterizedType) type; 
      if (expectedParameterizedType.equals(parameterizedType.getRawType())) { 
       final Type[] actualTypeArguments = parameterizedType.getActualTypeArguments(); 
       if (actualTypeArguments.length == 1) { 
        return actualTypeArguments[0]; 
       } 
      } 
     } 
     throw new IllegalArgumentException(String.valueOf(type)); 
    } 

} 

И ниже код будет использовать следующие закрывающиеся итераторов утилиты для того, чтобы закрыть ввода/вывода ресурсов (помните конвертер завод, который создает конвертер, который не закрывает ресурсы для итераторов?).

final class CloseableIterators { 

    private CloseableIterators() { 
    } 

    static <T> void forEachAndClose(final Iterator<? extends T> iterator, final Consumer<? super T> consumer) 
      throws Exception { 
     try { 
      while (iterator.hasNext()) { 
       consumer.accept(iterator.next()); 
      } 
     } finally { 
      tryClose(iterator); 
     } 
    } 

    static <T> List<T> collectToListAndClose(final Iterator<? extends T> iterator) 
      throws Exception { 
     final List<T> list = new ArrayList<>(); 
     forEachAndClose(iterator, list::add); 
     return unmodifiableList(list); 
    } 

    static void tryClose(final Object object) 
      throws Exception { 
     if (object instanceof AutoCloseable) { 
      ((AutoCloseable) object).close(); 
     } 
    } 

} 

Как вы можете видеть, все вышеприведенные методы пытаются закрыть заданные итераторы для освобождения ресурсов ввода-вывода. А вот как все это работает:

final Gson gson = new GsonBuilder() 
     .registerTypeAdapterFactory(getIteratorTypeAdapterFactory()) 
     .create(); 
final Retrofit retrofit = new Retrofit.Builder() 
     .baseUrl("http://localhost") 
     .addConverterFactory(getCustomGsonConverterFactory(gson)) 
     .build(); 
final IService service = retrofit.create(IService.class); 
final Call<Iterator<String>> getCall = service.get(); 
getCall.enqueue(new Callback<Iterator<String>>() { 
    @Override 
    public void onResponse(final Call<Iterator<String>> call, final Response<Iterator<String>> response) { 
     try { 
      final Iterator<String> iterator = response.body(); 
      if (...) { 
       forEachAndClose(iterator, out::println); 
      } else if (...) { 
       out.println(collectToListAndClose(iterator)); 
      } else { 
       tryClose(iterator); 
      } 
     } catch (final Exception ex) { 
      throw new RuntimeException(ex); 
     } 
    } 

    @Override 
    public void onFailure(final Call<Iterator<String>> call, final Throwable throwable) { 
     throwable.printStackTrace(err); 
    } 
}); 

Обратите внимание, что onResponseдолжен закрыть самый верхний Итератор объектов себя, или конкретный запрос ресурсов будет утечка: даже если нет никакого взаимодействия ожидается, tryClose сусло вызывается, чтобы закрыть экземпляр JsonReaderIterator. Но, как работают итераторы, вы можете использовать их один раз.

Edit: закрывающиеся итераторы повышение

Здесь у вас есть я думаю, что по крайней мере 2 способа, чтобы сделать их более надежными. Рассмотрим итератора, который может быть закрыт

interface IAutoCloseableIterator<E> 
     implements Iterator<E>, AutoCloseable { 
} 

Этот интерфейс может быть использован для вышеупомянутого JsonReaderIterator. И тогда вы можете:

• либо возвращать свои экземпляры через обратные вызовы, как Call<IAutoCloseableIterator<E>>;
• или обернуть любой итератор для того, чтобы обнаружить, если это AutoCloseable:

static <E, I extends Iterator<E> & AutoCloseable> I asAutoCloseable(final Iterator<E> iterator) { 
    final Iterator<E> resultIterator; 
    if (iterator instanceof AutoCloseable) { 
     resultIterator = iterator; 
    } else { 
     resultIterator = new IAutoCloseableIterator<E>() { 
      @Override 
      public boolean hasNext() { 
       return iterator.hasNext(); 
      } 

      @Override 
      public E next() { 
       return iterator.next(); 
      } 

      @Override 
      public void close() { 
       // do nothing or whatever elsewhere if there is no way to implement the method 
      } 

     }; 
    } 
    @SuppressWarnings("unchecked") 
    final I castIterator = (I) resultIterator; 
    return castIterator; 
} 

А затем (для первого варианта):

try (final IAutoCloseableIterator<String> iterator = response.body()) { 
    while (iterator.hasNext()) { 
     out.println(iterator.next()); 
    } 
} 

Или (для 2-го варианта, может быть не что ...):

try (final IAutoCloseableIterator<String> iterator = asAutoCloseable(response.body())) { 
    while (iterator.hasNext()) { 
     out.println(iterator.next()); 
    } 
} 

try-with-resources Ваш друг здесь.