(self, viewId)
| 760 | # RpcName: addViewObserver => viewport.geometry.view.observer.add |
| 761 | @exportRpc("viewport.geometry.view.observer.add") |
| 762 | def addViewObserver(self, viewId): |
| 763 | sView = self.getView(viewId) |
| 764 | if not sView: |
| 765 | return {"error": "Unable to get view with id %s" % viewId} |
| 766 | |
| 767 | realViewId = self.getApplication().GetObjectIdMap().GetGlobalId(sView) |
| 768 | |
| 769 | def pushGeometry(newSubscription=False): |
| 770 | stateToReturn = self.getViewState(realViewId, newSubscription) |
| 771 | stateToReturn["mtime"] = 0 if newSubscription else self.mtime |
| 772 | self.mtime += 1 |
| 773 | return stateToReturn |
| 774 | |
| 775 | if not realViewId in self.trackingViews: |
| 776 | observerCallback = lambda *args, **kwargs: self.publish( |
| 777 | "viewport.geometry.view.subscription", pushGeometry() |
| 778 | ) |
| 779 | tag = self.getApplication().AddObserver("UpdateEvent", observerCallback) |
| 780 | self.trackingViews[realViewId] = {"tags": [tag], "observerCount": 1} |
| 781 | else: |
| 782 | # There is an observer on this view already |
| 783 | self.trackingViews[realViewId]["observerCount"] += 1 |
| 784 | |
| 785 | self.publish("viewport.geometry.view.subscription", pushGeometry(True)) |
| 786 | return {"success": True, "viewId": realViewId} |
| 787 | |
| 788 | # RpcName: removeViewObserver => viewport.geometry.view.observer.remove |
| 789 | @exportRpc("viewport.geometry.view.observer.remove") |
nothing calls this directly
no test coverage detected