Merge pull request #6 from vzhn/feature/play_on_repeat

Feature/play on repeat

README.md

@@ -55,6 +55,7 @@ Value in a column ```errors``` increments when server is unable to send data chu
 | network.limits.size | bytes limit | 131072 bytes|
 | network.limits.time | chunk length limit | 200 ms |
 | streaming | stream  configuration | 2 sources: from filesystem (*.mkv) and from generated video |
+| streaming.file.repeat | true if repeat video | Generated |
 | streaming.file.class | filesystem source java class | Generated |
 | streaming.file.conf.basedir | default directory | ${application.directory}\video |
 | streaming.file.conf.file | default video file | jellyfish-5-mbps-hd-h264.mkv 
@@ -85,6 +86,7 @@ streaming:
   file:
     class: Filesystem
     conf:
+      repeat: true
       basedir: video
       file: jellyfish-5-mbps-hd-h264.mkv
   picture:

src/main/java/me/vzhilin/bstreamer/server/RtspServer.java

@@ -7,6 +7,8 @@
 import io.netty.channel.WriteBufferWaterMark;
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.kqueue.KQueueEventLoopGroup;
+import io.netty.channel.kqueue.KQueueServerSocketChannel;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
@@ -47,6 +49,10 @@ public RtspServer(Config serverConfig) {
             bossGroup = new NioEventLoopGroup(1);
             workerGroup = new NioEventLoopGroup(nThreads);
             channelClazz = NioServerSocketChannel.class;
+        } else if (AppRuntime.IS_MAC) {
+            bossGroup = new KQueueEventLoopGroup(1);
+            workerGroup = new KQueueEventLoopGroup(nThreads);
+            channelClazz = KQueueServerSocketChannel.class;
         } else {
             bossGroup = new EpollEventLoopGroup(1);
             workerGroup = new EpollEventLoopGroup(nThreads);

src/main/java/me/vzhilin/bstreamer/server/media/impl/PullSourceRegistry.java

@@ -4,6 +4,7 @@
 import me.vzhilin.bstreamer.server.SourceKey;
 import me.vzhilin.bstreamer.server.scheduler.BufferingLimits;
 import me.vzhilin.bstreamer.server.scheduler.PushSource;
+import me.vzhilin.bstreamer.server.streaming.RepeatedSource;
 import me.vzhilin.bstreamer.server.streaming.base.PullSource;
 import me.vzhilin.bstreamer.util.PropertyMap;
 
@@ -34,13 +35,19 @@ private Supplier<PullSource> supplierFor(SourceKey key) {
         try {
             Class<PullSource> pullSource = (Class<PullSource>) Class.forName(DEFAULT_PACKAGE + key.clazz);
             Constructor<PullSource> constructor = pullSource.getDeclaredConstructor(ServerContext.class, PropertyMap.class);
-            return () -> {
+            Supplier<PullSource> pullSourceSupplier = () -> {
                 try {
                     return constructor.newInstance(serverContext, key.cfg);
                 } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
                     throw new RuntimeException(e);
                 }
             };
+
+            if (key.cfg.getBoolean("repeat")) {
+                return () -> new RepeatedSource(pullSourceSupplier);
+            } else {
+                return pullSourceSupplier;
+            }
         } catch (ClassNotFoundException | NoSuchMethodException e) {
             throw new RuntimeException(e);
         }

src/main/java/me/vzhilin/bstreamer/server/streaming/RepeatedSource.java

@@ -0,0 +1,67 @@
+package me.vzhilin.bstreamer.server.streaming;
+
+import me.vzhilin.bstreamer.server.streaming.base.PullSource;
+import me.vzhilin.bstreamer.server.streaming.file.MediaPacket;
+import me.vzhilin.bstreamer.server.streaming.file.SourceDescription;
+
+import java.io.IOException;
+import java.util.function.Supplier;
+
+public class RepeatedSource implements PullSource {
+  private final Supplier<PullSource> supplier;
+  private PullSource delegate = null;
+  private long lastDts;
+  private long lastPts;
+  private long ptsOffset = 0;
+  private long dtsOffset = 0;
+
+  public RepeatedSource(Supplier<PullSource> supplier) {
+    this.supplier = supplier;
+  }
+
+  private PullSource ensureHasDelegate() {
+    if (delegate == null) {
+      delegate = supplier.get();
+    }
+    return delegate;
+  }
+  @Override
+  public SourceDescription getDesc() {
+    ensureHasDelegate();
+    return delegate.getDesc();
+  }
+
+  @Override
+  public boolean hasNext() {
+    ensureHasDelegate();
+
+    if (!delegate.hasNext()) {
+      ptsOffset += lastPts;
+      dtsOffset += lastDts;
+      try {
+        delegate.close();
+      } catch (IOException e) {
+        return false;
+      }
+      delegate = supplier.get();
+    }
+
+    return delegate.hasNext();
+  }
+
+  @Override
+  public MediaPacket next() {
+    MediaPacket p = delegate.next();
+    lastDts = p.getDts();
+    lastPts = p.getPts();
+    return new MediaPacket(p.getPts() + ptsOffset, p.getDts() + dtsOffset, p.isKey(), p.getPayload());
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (delegate != null) {
+      delegate.close();
+      delegate = null;
+    }
+  }
+}

src/main/java/me/vzhilin/bstreamer/util/AppRuntime.java

@@ -6,6 +6,7 @@
 
 public final class AppRuntime {
     public final static boolean IS_WINDOWS = System.getProperty("os.name").toLowerCase().contains("win");
+    public final static boolean IS_MAC = System.getProperty("os.name").toLowerCase().contains("mac");
     public final static File APP_PATH = getApplcatonPath();
 
     private AppRuntime() { }

src/main/java/me/vzhilin/bstreamer/util/PropertyMap.java

@@ -79,6 +79,10 @@ public String getString(String key) {
         return (String) getObject(key);
     }
 
+    public Boolean getBoolean(String key) {
+        return Boolean.valueOf(getString(key, "false"));
+    }
+
     public List<String> getStringArray(String key) {
         return (List<String>) getObject(key);
     }