1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187 |
--- ClientBroadcastStream.java
+++ ClientBroadcastStream.java
@@ -96,13 +96,13 @@
/**
* Logger
*/
private static final Logger log = LoggerFactory.getLogger(ClientBroadcastStream.class);
/** Stores absolute time for video stream. */
- private int audioTime = -1;
+ private int audioTime = 0;
/**
* Total number of bytes received.
*/
private long bytesReceived;
@@ -129,13 +129,13 @@
/**
* Timestamp the stream was created.
*/
private long creationTime;
/** Stores absolute time for data stream. */
- private int dataTime = -1;
+ private int dataTime = 0;
/** Stores timestamp of first packet. */
private int firstPacketTime = -1;
/**
* Pipe for live streaming
@@ -185,13 +185,16 @@
/**
* Factory object for video codecs
*/
private VideoCodecFactory videoCodecFactory = null;
/** Stores absolute time for audio stream. */
- private int videoTime = -1;
+ private int videoTime = 0;
+
+ private int minStreamTime = 0;
+ public void setMinStreamTime(int mst) { minStreamTime = mst; }
/** Listeners to get notified about received packets. */
private Set<IStreamListener> listeners = new CopyOnWriteArraySet<IStreamListener>();
/**
* Check and send notification if necessary
@@ -253,21 +256,50 @@
rtmpEvent = (IRTMPEvent) event;
} catch (ClassCastException e) {
log.error("Class cast exception in event dispatch", e);
return;
}
int eventTime = -1;
- // If this is first packet save it's timestamp
- if (firstPacketTime == -1) {
- firstPacketTime = rtmpEvent.getTimestamp();
+ if (log.isDebugEnabled()) {
+ // If this is first packet save its timestamp; expect it is absolute? no matter: it's never used!
+ if (firstPacketTime == -1) {
+ firstPacketTime = rtmpEvent.getTimestamp();
+ log.debug(String.format("CBS=@%08x: firstPacketTime=%d %s"
+ ,System.identityHashCode(this)
+ ,firstPacketTime
+ ,(rtmpEvent.getHeader().isTimerRelative()?"(rel)":"(abs)")
+ ));
+ }
+
+ int basetime = ((rtmpEvent instanceof VideoData) ? videoTime
+ : ((rtmpEvent instanceof AudioData) ? audioTime : dataTime));
+ if (rtmpEvent.getHeader().isTimerRelative()) {
+ int abstime = rtmpEvent.getTimestamp() + basetime;
+ log.debug(String.format("CBS=@%08x: rtmpEvent=%s timestamp=%d (relative)+%d = %d"
+ ,System.identityHashCode(this)
+ ,rtmpEvent.getClass().getSimpleName()
+ ,rtmpEvent.getTimestamp()
+ ,basetime
+ ,abstime));
+ } else {
+ int deltime = rtmpEvent.getTimestamp() - basetime;
+ log.debug(String.format("CBS=@%08x: rtmpEvent=%s timestamp=%d (absolute)-%d = %d"
+ ,System.identityHashCode(this)
+ ,rtmpEvent.getClass().getSimpleName()
+ ,rtmpEvent.getTimestamp()
+ ,basetime
+ ,deltime));
+ }
}
if (rtmpEvent instanceof AudioData) {
if (info != null) {
info.setHasAudio(true);
}
if (rtmpEvent.getHeader().isTimerRelative()) {
+ if (audioTime == 0)
+ log.warn("First Audio timestamp is relative!"+rtmpEvent.getTimestamp());
audioTime += rtmpEvent.getTimestamp();
} else {
audioTime = rtmpEvent.getTimestamp();
}
eventTime = audioTime;
} else if (rtmpEvent instanceof VideoData) {
@@ -289,26 +321,50 @@
}
if (info != null) {
info.setHasVideo(true);
}
if (rtmpEvent.getHeader().isTimerRelative()) {
+ if (videoTime == 0)
+ log.warn("First Video timestamp is relative!"+rtmpEvent.getTimestamp());
videoTime += rtmpEvent.getTimestamp();
} else {
videoTime = rtmpEvent.getTimestamp();
+ // Flash player may send first VideoData with old-absolute timestamp.
+ // This ruins the stream's timebase in FileConsumer.
+ // We don't want to discard the packet, as it may be a video keyframe.
+ // Generally a Data or Audio packet has set the timebase to a reasonable value,
+ // Eventually a new/correct absolute time will come on the video channel.
+ // We could put this logic between livePipe and filePipe;
+ // This would work for Audio Data as well, but have not seen the need.
+ int cts = Math.max(audioTime, dataTime);
+ cts = Math.max(cts, minStreamTime);
+ int fudge = 20;
+ // accept some slightly (20ms) retro timestamps [this may not be needed,
+ // the publish Data should strictly precede the video data]
+ if (videoTime + fudge < cts) {
+ if (log.isInfoEnabled()) {
+ log.info(String.format("dispatchEvent: adjust archaic videoTime, from: %d to %d", videoTime, cts));
+ }
+ videoTime = cts;
+ }
}
eventTime = videoTime;
} else if (rtmpEvent instanceof Invoke) {
if (rtmpEvent.getHeader().isTimerRelative()) {
+ if (dataTime < 0)
+ log.warn("First data [Invoke] timestamp is relative!"+rtmpEvent.getTimestamp());
dataTime += rtmpEvent.getTimestamp();
} else {
dataTime = rtmpEvent.getTimestamp();
}
return;
} else if (rtmpEvent instanceof Notify) {
if (rtmpEvent.getHeader().isTimerRelative()) {
+ if (dataTime == 0)
+ log.warn("First data [Notify] timestamp is relative!"+rtmpEvent.getTimestamp());
dataTime += rtmpEvent.getTimestamp();
} else {
dataTime = rtmpEvent.getTimestamp();
}
eventTime = dataTime;
}
@@ -321,13 +377,14 @@
// Notify event listeners
checkSendNotifications(event);
// Create new RTMP message, initialize it and push through pipe
RTMPMessage msg = new RTMPMessage();
msg.setBody(rtmpEvent);
- msg.getBody().setTimestamp(eventTime);
+ msg.getBody().setTimestamp(eventTime); // rtmpEvent.setTimestamp(eventTime); ~ABSOLUTE!
+ // note this timestamp is set in event/body but not in the associated header.
try {
if (livePipe != null) {
livePipe.pushMessage(msg);
}
recordPipe.pushMessage(msg);
} catch (IOException err) {
@@ -747,13 +804,14 @@
videoCodecFactory = (VideoCodecFactory) getScope().getContext()
.getBean(VideoCodecFactory.KEY);
checkVideoCodec = true;
} catch (Exception err) {
log.warn("No video codec factory available.", err);
}
- firstPacketTime = audioTime = videoTime = dataTime = -1;
+ firstPacketTime = -1;
+ audioTime = videoTime = dataTime = 0;
connMsgOut = consumerManager.getConsumerOutput(this);
connMsgOut.subscribe(this, null);
recordPipe = new InMemoryPushPushPipe();
Map<Object, Object> recordParamMap = new HashMap<Object, Object>();
// Clear record flag
recordParamMap.put("record", null); |