ambiser
/
openvidu
mirror of https://github.com/OpenVidu/openvidu.git
8
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

openvidu-server: blocking and non-blocking versions of endpoint connection

pull/621/head
pabloFuente 2021-03-12 11:25:13 +01:00
parent 17ea55ad07
commit 2e40d14432
4 changed files with 106 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
openvidu-server/src/main/java/io/openvidu/server/kurento/endpoint/PublisherEndpoint.java
View File

@ -175,45 +175,42 @@ public class PublisherEndpoint extends MediaEndpoint {
/**
* Initializes this media endpoint for publishing media and processes the SDP
* offer or answer. If the internal endpoint is an {@link WebRtcEndpoint}, it
* first registers an event listener for the ICE candidates and instructs the
* endpoint to start gathering the candidates. If required, it connects to
* itself (after applying the intermediate media elements and the
* {@link PassThrough}) to allow loopback of the media stream.
* offer. If the internal endpoint is an {@link WebRtcEndpoint}, it first
* registers an event listener for the ICE candidates and instructs the endpoint
* to start gathering the candidates. If required, it connects to itself (after
* applying the intermediate media elements and the {@link PassThrough}) to
* allow loopback of the media stream.
*
* @param sdpOffer offer from the remote peer
* @param sdpOffer SDP offer from the remote peer
* @param doLoopback loopback flag
* @param loopbackAlternativeSrc alternative loopback source
* @param loopbackConnectionType how to connect the loopback source
* @return the SDP response (the answer if processing an offer SDP, otherwise is
* the updated offer generated previously by this endpoint)
* @return the SDP answer
*/
public synchronized String publish(String sdpOffer, boolean doLoopback) {
String sdpResponse = processOffer(sdpOffer);
registerOnIceCandidateEventListener(this.getOwner().getParticipantPublicId());
if (doLoopback) {
connect(this.getEndpoint());
connect(this.getEndpoint(), false);
} else {
innerConnect();
innerConnect(false);
}
this.createdAt = System.currentTimeMillis();
String sdpResponse = processOffer(sdpOffer);
gatherCandidates();
return sdpResponse;
}
public synchronized void connect(MediaElement sink) {
public synchronized void connect(MediaElement sink, boolean blocking) {
if (!connected) {
innerConnect();
innerConnect(blocking);
}
internalSinkConnect(passThru, sink);
internalSinkConnect(passThru, sink, blocking);
this.enableIpCameraIfNecessary();
}
public synchronized void connect(MediaElement sink, MediaType type) {
public synchronized void connect(MediaElement sink, MediaType type, boolean blocking) {
if (!connected) {
innerConnect();
innerConnect(blocking);
}
internalSinkConnect(passThru, sink, type);
internalSinkConnect(passThru, sink, type, blocking);
this.enableIpCameraIfNecessary();
}
@ -231,7 +228,7 @@ public class PublisherEndpoint extends MediaEndpoint {
}
public synchronized void disconnectFrom(MediaElement sink) {
internalSinkDisconnect(passThru, sink);
internalSinkDisconnect(passThru, sink, false);
}
/**
@ -277,11 +274,11 @@ public class PublisherEndpoint extends MediaEndpoint {
}
if (connected) {
if (first != null) {
internalSinkConnect(first, shaper, type);
internalSinkConnect(first, shaper, type, false);
} else {
internalSinkConnect(this.getEndpoint(), shaper, type);
internalSinkConnect(this.getEndpoint(), shaper, type, false);
}
internalSinkConnect(shaper, passThru, type);
internalSinkConnect(shaper, passThru, type, false);
}
elementIds.addFirst(id);
elements.put(id, shaper);
@ -331,7 +328,7 @@ public class PublisherEndpoint extends MediaEndpoint {
} else {
prev = passThru;
}
internalSinkConnect(next, prev);
internalSinkConnect(next, prev, false);
}
elementIds.remove(elementId);
if (releaseElement) {
@ -370,13 +367,13 @@ public class PublisherEndpoint extends MediaEndpoint {
}
switch (muteType) {
case ALL:
internalSinkDisconnect(this.getEndpoint(), sink);
internalSinkDisconnect(this.getEndpoint(), sink, false);
break;
case AUDIO:
internalSinkDisconnect(this.getEndpoint(), sink, MediaType.AUDIO);
internalSinkDisconnect(this.getEndpoint(), sink, MediaType.AUDIO, false);
break;
case VIDEO:
internalSinkDisconnect(this.getEndpoint(), sink, MediaType.VIDEO);
internalSinkDisconnect(this.getEndpoint(), sink, MediaType.VIDEO, false);
break;
}
}
@ -396,19 +393,19 @@ public class PublisherEndpoint extends MediaEndpoint {
}
switch (muteType) {
case ALL:
internalSinkConnect(this.getEndpoint(), sink);
internalSinkConnect(this.getEndpoint(), sink, false);
break;
case AUDIO:
internalSinkConnect(this.getEndpoint(), sink, MediaType.AUDIO);
internalSinkConnect(this.getEndpoint(), sink, MediaType.AUDIO, false);
break;
case VIDEO:
internalSinkConnect(this.getEndpoint(), sink, MediaType.VIDEO);
internalSinkConnect(this.getEndpoint(), sink, MediaType.VIDEO, false);
break;
}
}
public synchronized PassThrough disconnectFromPassThrough() {
this.internalSinkDisconnect(this.getWebEndpoint(), this.passThru);
this.internalSinkDisconnect(this.getWebEndpoint(), this.passThru, false);
return this.passThru;
}
@ -428,7 +425,7 @@ public class PublisherEndpoint extends MediaEndpoint {
return elementIds.get(idx - 1);
}
private void innerConnect() {
private void innerConnect(boolean blocking) {
if (this.getEndpoint() == null) {
throw new OpenViduException(Code.MEDIA_ENDPOINT_ERROR_CODE,
"Can't connect null endpoint (ep: " + getEndpointName() + ")");
@ -441,15 +438,18 @@ public class PublisherEndpoint extends MediaEndpoint {
throw new OpenViduException(Code.MEDIA_ENDPOINT_ERROR_CODE,
"No media element with id " + prevId + " (ep: " + getEndpointName() + ")");
}
internalSinkConnect(current, prev);
internalSinkConnect(current, prev, blocking);
current = prev;
prevId = getPrevious(prevId);
}
internalSinkConnect(current, passThru);
internalSinkConnect(current, passThru, blocking);
connected = true;
}
private void internalSinkConnect(final MediaElement source, final MediaElement sink) {
private void internalSinkConnect(final MediaElement source, final MediaElement sink, boolean blocking) {
if (blocking) {
source.connect(sink);
} else {
source.connect(sink, new Continuation<Void>() {
@Override
public void onSuccess(Void result) throws Exception {
@ -464,6 +464,7 @@ public class PublisherEndpoint extends MediaEndpoint {
}
});
}
}
/**
* Same as {@link #internalSinkConnect(MediaElement, MediaElement)}, but can
@ -476,15 +477,19 @@ public class PublisherEndpoint extends MediaEndpoint {
* be used instead
* @see #internalSinkConnect(MediaElement, MediaElement)
*/
private void internalSinkConnect(final MediaElement source, final MediaElement sink, final MediaType type) {
private void internalSinkConnect(final MediaElement source, final MediaElement sink, final MediaType type,
boolean blocking) {
if (type == null) {
internalSinkConnect(source, sink);
internalSinkConnect(source, sink, blocking);
} else {
if (blocking) {
source.connect(sink, type);
} else {
source.connect(sink, type, new Continuation<Void>() {
@Override
public void onSuccess(Void result) throws Exception {
log.debug("EP {}: {} media elements have been connected (source {} -> sink {})", getEndpointName(),
type, source.getId(), sink.getId());
log.debug("EP {}: {} media elements have been connected (source {} -> sink {})",
getEndpointName(), type, source.getId(), sink.getId());
}
@Override
@ -495,8 +500,12 @@ public class PublisherEndpoint extends MediaEndpoint {
});
}
}
}
private void internalSinkDisconnect(final MediaElement source, final MediaElement sink) {
private void internalSinkDisconnect(final MediaElement source, final MediaElement sink, boolean blocking) {
if (blocking) {
source.disconnect(sink);
} else {
source.disconnect(sink, new Continuation<Void>() {
@Override
public void onSuccess(Void result) throws Exception {
@ -511,6 +520,7 @@ public class PublisherEndpoint extends MediaEndpoint {
}
});
}
}
/**
* Same as {@link #internalSinkDisconnect(MediaElement, MediaElement)}, but can
@ -523,9 +533,13 @@ public class PublisherEndpoint extends MediaEndpoint {
* be used instead
* @see #internalSinkConnect(MediaElement, MediaElement)
*/
private void internalSinkDisconnect(final MediaElement source, final MediaElement sink, final MediaType type) {
private void internalSinkDisconnect(final MediaElement source, final MediaElement sink, final MediaType type,
boolean blocking) {
if (type == null) {
internalSinkDisconnect(source, sink);
internalSinkDisconnect(source, sink, blocking);
} else {
if (blocking) {
source.disconnect(sink, type);
} else {
source.disconnect(sink, type, new Continuation<Void>() {
@Override
@ -536,12 +550,13 @@ public class PublisherEndpoint extends MediaEndpoint {
@Override
public void onError(Throwable cause) throws Exception {
log.warn("EP {}: Failed to disconnect {} media elements (source {} -> sink {})", getEndpointName(),
type, source.getId(), sink.getId(), cause);
log.warn("EP {}: Failed to disconnect {} media elements (source {} -> sink {})",
getEndpointName(), type, source.getId(), sink.getId(), cause);
}
});
}
}
}
public MediaOptions getMediaOptions() {
return mediaOptions;

2
openvidu-server/src/main/java/io/openvidu/server/kurento/endpoint/SubscriberEndpoint.java
View File

@ -50,7 +50,7 @@ public class SubscriberEndpoint extends MediaEndpoint {
this.createdAt = System.currentTimeMillis();
String sdpAnswer = processOffer(sdpOffer);
gatherCandidates();
publisher.connect(this.getEndpoint());
publisher.connect(this.getEndpoint(), false);
this.publisherStreamId = publisher.getStreamId();
return sdpAnswer;
}

2
openvidu-server/src/main/java/io/openvidu/server/recording/CompositeWrapper.java
View File

@ -111,7 +111,7 @@ public class CompositeWrapper {
public void connectPublisherEndpoint(PublisherEndpoint endpoint) throws OpenViduException {
HubPort hubPort = new HubPort.Builder(composite).build();
endpoint.connect(hubPort);
endpoint.connect(hubPort, false);
String streamId = endpoint.getOwner().getPublisherStreamId();
this.hubPorts.put(streamId, hubPort);
this.publisherEndpoints.put(streamId, endpoint);

8
openvidu-server/src/main/java/io/openvidu/server/recording/service/SingleStreamRecordingService.java
View File

@ -392,14 +392,14 @@ public class SingleStreamRecordingService extends RecordingService {
MediaProfileSpecType profile) {
switch (profile) {
case WEBM:
publisherEndpoint.connect(recorder, MediaType.AUDIO);
publisherEndpoint.connect(recorder, MediaType.VIDEO);
publisherEndpoint.connect(recorder, MediaType.AUDIO, false);
publisherEndpoint.connect(recorder, MediaType.VIDEO, false);
break;
case WEBM_AUDIO_ONLY:
publisherEndpoint.connect(recorder, MediaType.AUDIO);
publisherEndpoint.connect(recorder, MediaType.AUDIO, false);
break;
case WEBM_VIDEO_ONLY:
publisherEndpoint.connect(recorder, MediaType.VIDEO);
publisherEndpoint.connect(recorder, MediaType.VIDEO, false);
break;
default:
throw new UnsupportedOperationException("Unsupported profile when single stream recording: " + profile);