package cn.org.hentai.jtt1078.subscriber;
|
|
import cn.org.hentai.jtt1078.flv.FlvEncoder;
|
import cn.org.hentai.jtt1078.util.Packet;
|
import io.netty.channel.ChannelFuture;
|
import io.netty.channel.ChannelHandlerContext;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
|
import java.util.LinkedList;
|
import java.util.concurrent.atomic.AtomicLong;
|
|
/**
|
* Created by matrixy on 2020/1/11.
|
*/
|
public abstract class Subscriber extends Thread
|
{
|
static Logger logger = LoggerFactory.getLogger(Subscriber.class);
|
static final AtomicLong SEQUENCE = new AtomicLong(0L);
|
|
private long id;
|
private String tag;
|
private Object lock;
|
private ChannelHandlerContext context;
|
protected LinkedList<byte[]> messages;
|
|
public Subscriber(String tag, ChannelHandlerContext ctx)
|
{
|
this.tag = tag;
|
this.context = ctx;
|
this.lock = new Object();
|
this.messages = new LinkedList<byte[]>();
|
|
this.id = SEQUENCE.getAndAdd(1L);
|
}
|
|
public long getId()
|
{
|
return this.id;
|
}
|
|
public String getTag()
|
{
|
return this.tag;
|
}
|
|
public abstract void onVideoData(long timeoffset, byte[] data, FlvEncoder flvEncoder);
|
|
public abstract void onAudioData(long timeoffset, byte[] data, FlvEncoder flvEncoder);
|
|
public void enqueue(byte[] data)
|
{
|
if (data == null) return;
|
synchronized (lock)
|
{
|
messages.addLast(data);
|
lock.notify();
|
}
|
}
|
|
public void run()
|
{
|
loop : while (!this.isInterrupted())
|
{
|
try
|
{
|
byte[] data = take();
|
if (data != null) send(data).await();
|
}
|
catch(Exception ex)
|
{
|
//销毁线程时，如果有锁wait就不会销毁线程，抛出InterruptedException异常
|
if (ex instanceof InterruptedException)
|
{
|
break loop;
|
}
|
logger.error("send failed", ex);
|
}
|
}
|
logger.info("subscriber closed");
|
}
|
|
protected byte[] take()
|
{
|
byte[] data = null;
|
try
|
{
|
synchronized (lock)
|
{
|
while (messages.isEmpty())
|
{
|
lock.wait(100);
|
if (this.isInterrupted()) return null;
|
}
|
data = messages.removeFirst();
|
}
|
return data;
|
}
|
catch(Exception ex)
|
{
|
this.interrupt();
|
return null;
|
}
|
}
|
|
public void close()
|
{
|
this.interrupt();
|
}
|
|
public ChannelFuture send(byte[] message)
|
{
|
return context.writeAndFlush(message);
|
}
|
}
|
