001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.juneau.marshall.sse;
018
019import static org.apache.juneau.commons.utils.Shorts.*;
020import static org.apache.juneau.commons.utils.StringUtils.*;
021
022import java.io.*;
023import java.util.*;
024import java.util.stream.*;
025
026import org.apache.juneau.marshall.serializer.*;
027import org.apache.juneau.marshall.stream.*;
028
029/**
030 * Session object that lives for the duration of a single use of {@link SseSerializer}.
031 *
032 * <h5 class='section'>Notes:</h5><ul>
033 *    <li class='warn'>This class is not thread safe and is typically discarded after one use.
034 * </ul>
035 *
036 * <h5 class='section'>See Also:</h5><ul>
037 *    <li class='link'><a class="doclink" href="https://juneau.apache.org/docs/topics/Sse">SSE</a>
038 * </ul>
039 */
040@SuppressWarnings({
041   "java:S110", // Inheritance depth acceptable for session hierarchy
042   "java:S1192", // Duplicated literals (argument/property names) read more clearly inline than as constants
043   "resource" // Closeable resources are owned by the caller's serializer session; Eclipse JDT @Owning warning is by design.
044})
045public class SseSerializerSession extends WriterSerializerSession implements RecordWritable {
046
047   /**
048    * Builder class.
049    */
050   public static class Builder extends WriterSerializerSession.Builder<Builder> {
051
052      /**
053       * Constructor.
054       *
055       * @param ctx The context creating this session. Cannot be <jk>null</jk>.
056       */
057      protected Builder(SseSerializer ctx) {
058         super(reqnn("ctx", ctx));
059      }
060
061      @Override
062      public SseSerializerSession build() {
063         return new SseSerializerSession(this);
064      }
065   }
066
067   /**
068    * Creates a new builder for this object.
069    *
070    * @param ctx The context creating this session. Cannot be <jk>null</jk>.
071    * @return A new builder.
072    */
073   public static Builder create(SseSerializer ctx) {
074      return new Builder(reqnn("ctx", ctx));
075   }
076
077   /**
078    * Constructor.
079    *
080    * @param builder The builder for this object.
081    */
082   protected SseSerializerSession(Builder builder) {
083      super(builder);
084   }
085
086   @Override /* RecordWritable */
087   public RecordWriter writeRecords(Object output) throws IOException {
088      return RecordAdapter.arrayWriter(this, output);
089   }
090
091   @Override /* RecordWritable */
092   public boolean isRecordStreaming() {
093      return false;
094   }
095
096   @Override /* Overridden from SerializerSession */
097   @SuppressWarnings({
098      "java:S2095" // Writer is owned by SerializerPipe/caller and must not be closed by this session.
099   })
100   protected void doWrite(SerializerPipe pipe, Object o) throws IOException, SerializeException {
101      var w = pipe.getWriter();
102      if (o == null)
103         return;
104      if (o instanceof SseEvent o2) {
105         writeEvent(w, o2);
106      } else if (o instanceof Iterable<?> o2) {
107         for (var item : o2)
108            writeIfEvent(w, item);
109      } else if (o instanceof Stream<?> o2) {
110         try (var s2 = o2) {
111            s2.forEach(item -> writeIfEventUnchecked(w, item));
112         }
113      } else if (o instanceof SseEvent[] o2) {
114         for (var e : o2)
115            writeEvent(w, e);
116      } else if (o.getClass().isArray()) {
117         var arr = (Object[]) o;
118         for (var item : arr)
119            writeIfEvent(w, item);
120      } else {
121         throw new SerializeException("SseSerializer cannot serialize objects of type '%s'.", cn(o));
122      }
123   }
124
125   private static void writeIfEvent(Writer w, Object item) throws IOException, SerializeException {
126      if (item == null)
127         return;
128      if (!(item instanceof SseEvent item2))
129         throw new SerializeException("SseSerializer encountered non-SseEvent element of type '%s'.", cn(item));
130      writeEvent(w, item2);
131   }
132
133   private static void writeIfEventUnchecked(Writer w, Object item) {
134      try {
135         writeIfEvent(w, item);
136      } catch (IOException e) {
137         throw new UncheckedIOException(e);
138      } catch (SerializeException e) {
139         throw new IllegalArgumentException(e);
140      }
141   }
142
143   private static void writeEvent(Writer w, SseEvent e) throws IOException {
144      if (e == null)
145         return;
146      if (!isEmpty(e.getEvent()))
147         writeField(w, "event", e.getEvent());
148      if (e.getData() != null) {
149         for (var line : splitData(e.getData()))
150            writeField(w, "data", line);
151      }
152      if (e.getId() != null)
153         writeField(w, "id", e.getId());
154      if (e.getRetry() != null)
155         w.write("retry: " + e.getRetry() + "\n");
156      w.write("\n");
157      w.flush();
158   }
159
160   private static void writeField(Writer w, String name, String value) throws IOException {
161      w.write(name);
162      w.write(": ");
163      w.write(value);
164      w.write('\n');
165   }
166
167   static List<String> splitData(String s) {
168      var parts = s.split("\n", -1);
169      return Arrays.asList(parts);
170   }
171}