+11

.changeset/cuddly-actors-look.md

reviewed

··· 1 1 + --- 2 2 + '@urql/exchange-multipart-fetch': minor 3 3 + '@urql/exchange-graphcache': minor 4 4 + '@urql/exchange-persisted': minor 5 5 + '@urql/exchange-context': minor 6 6 + '@urql/exchange-execute': minor 7 7 + '@urql/exchange-retry': minor 8 8 + '@urql/exchange-auth': minor 9 9 + --- 10 10 + 11 11 + Update exchanges to drop redundant `share` calls, since `@urql/core`’s `composeExchanges` utility now automatically does so for us.

+5

.changeset/five-lies-collect.md

reviewed

··· 1 1 + --- 2 2 + '@urql/core': patch 3 3 + --- 4 4 + 5 5 + Update `Exchange` contract and `composeExchanges` utility to remove the need to manually call `share` on either incoming `Source<Operation>` or `forward()`’s `Source<OperationResult>`. This is now taken care of internally in `composeExchanges` and should make it easier for you to create custom exchanges and for us to explain them.

+22 -85

docs/advanced/authoring-exchanges.md

reviewed

··· 152 152 }); 153 153 ``` 154 154 155 155 - ### Only One Operations Stream 156 156 - 157 157 - When writing an Exchange we have to be careful not to _split_ the stream into multiple ones by 158 158 - subscribing multiple times. Streams are lazy and immutable by default. Every time you use them, 159 159 - a new chain of streaming operators is created; since Exchanges are technically side effects, we don't 160 160 - want to accidentally have multiple instances of them in parallel. 161 161 - 162 162 - The `ExchangeIO` function receives an `operations$` stream. It's important to be careful to either only 163 163 - use it once, or to _share_ its subscription. 164 164 - 165 165 - ```js 166 166 - import { pipe, filter, merge, share } from 'wonka'; 167 167 - 168 168 - // DON'T: split use operations$ twice 169 169 - ({ forward }) => operations$ => { 170 170 - // <-- The ExchangeIO function (inline) 171 171 - const queries = pipe( 172 172 - operations$, 173 173 - filter(op => op.kind === 'query') 174 174 - ); 175 175 - const others = pipe( 176 176 - operations$, 177 177 - filter(op => op.kind !== 'query') 178 178 - ); 179 179 - return forward(merge([queries, others])); 180 180 - }; 181 181 - 182 182 - // DO: share operations$ if you have to use it twice 183 183 - ({ forward }) => operations$ => { 184 184 - // <-- The ExchangeIO function (inline) 185 185 - const shared = pipe(operations$, share); 186 186 - const queries = pipe( 187 187 - shared, 188 188 - filter(op => op.kind === 'query') 189 189 - ); 190 190 - const others = pipe( 191 191 - shared, 192 192 - filter(op => op.kind !== 'query') 193 193 - ); 194 194 - return forward(merge([queries, others])); 195 195 - }; 196 196 - 197 197 - // DO: use operations$ only once alternatively 198 198 - ({ forward }) => ( 199 199 - operations$ // <-- The ExchangeIO function (inline) 200 200 - ) => 201 201 - pipe( 202 202 - operations$, 203 203 - map(op => { 204 204 - if (op.kind === 'query') { 205 205 - /* ... */ 206 206 - } else { 207 207 - /* ... */ 208 208 - } 209 209 - return op; 210 210 - }), 211 211 - forward 212 212 - ); 213 213 - ``` 214 214 - 215 215 - So if you see the `operations$` stream twice in your exchange code, make sure to 216 216 - use Wonka's [`share`](https://wonka.kitten.sh/api/operators#share) operator, to share the underlying 217 217 - subscription between all your streams. 218 218 - 219 155 ### How to Avoid Accidentally Dropping Operations 220 156 221 157 Typically the `operations$` stream will send you `query`, `mutation`, ··· 226 162 not `filter` operations too aggressively. 227 163 228 164 ```js 229 229 - import { pipe, filter, merge, share } from 'wonka'; 165 165 + import { pipe, filter, merge } from 'wonka'; 230 166 231 167 // DON'T: drop unknown operations 232 232 - ({ forward }) => operations$ => { 233 233 - // This doesn't handle operations that aren't queries 234 234 - const queries = pipe( 235 235 - operations$, 236 236 - filter(op => op.kind === 'query') 237 237 - ); 238 238 - return forward(queries); 239 239 - }; 168 168 + ({ forward }) => 169 169 + operations$ => { 170 170 + // This doesn't handle operations that aren't queries 171 171 + const queries = pipe( 172 172 + operations$, 173 173 + filter(op => op.kind === 'query') 174 174 + ); 175 175 + return forward(queries); 176 176 + }; 240 177 241 178 // DO: forward operations that you don't handle 242 242 - ({ forward }) => operations$ => { 243 243 - const shared = pipe(operations$, share); 244 244 - const queries = pipe( 245 245 - shared, 246 246 - filter(op => op.kind === 'query') 247 247 - ); 248 248 - const rest = pipe( 249 249 - shared, 250 250 - filter(op => op.kind !== 'query') 251 251 - ); 252 252 - return forward(merge([queries, rest])); 253 253 - }; 179 179 + ({ forward }) => 180 180 + operations$ => { 181 181 + const queries = pipe( 182 182 + operations$, 183 183 + filter(op => op.kind === 'query') 184 184 + ); 185 185 + const rest = pipe( 186 186 + operations$, 187 187 + filter(op => op.kind !== 'query') 188 188 + ); 189 189 + return forward(merge([queries, rest])); 190 190 + }; 254 191 ``` 255 192 256 193 If operations are grouped and/or filtered by what the exchange is handling, then it's also important to

+3 -1

exchanges/auth/src/authExchange.test.ts

reviewed

··· 5 5 toPromise, 6 6 take, 7 7 makeSubject, 8 8 + share, 8 9 publish, 9 10 scan, 10 11 tap, ··· 41 42 pipe( 42 43 op$, 43 44 tap(op => operations.push(op)), 44 44 - map(result) 45 45 + map(result), 46 46 + share 45 47 ), 46 48 client: new Client({ 47 49 url: '/api',

+3 -6

exchanges/auth/src/authExchange.ts

reviewed

··· 8 8 makeSubject, 9 9 toPromise, 10 10 merge, 11 11 - share, 12 11 } from 'wonka'; 13 12 14 13 import { ··· 300 299 return config ? config.addAuthToOperation(operation) : operation; 301 300 } 302 301 303 303 - const sharedOps$ = pipe(operations$, share); 304 304 - 305 302 const teardownOps$ = pipe( 306 306 - sharedOps$, 303 303 + operations$, 307 304 filter(operation => operation.kind === 'teardown') 308 305 ); 309 306 310 307 const pendingOps$ = pipe( 311 311 - sharedOps$, 308 308 + operations$, 312 309 filter(operation => operation.kind !== 'teardown') 313 310 ); 314 311 ··· 337 334 filter(Boolean) 338 335 ) as Source<Operation>; 339 336 340 340 - const result$ = pipe(merge([opsWithAuth$, teardownOps$]), forward, share); 337 337 + const result$ = pipe(merge([opsWithAuth$, teardownOps$]), forward); 341 338 342 339 return pipe( 343 340 result$,

+1

exchanges/context/src/context.ts

reviewed

··· 4 4 Operation, 5 5 OperationContext, 6 6 } from '@urql/core'; 7 7 + 7 8 import { fromPromise, fromValue, mergeMap, pipe } from 'wonka'; 8 9 9 10 /** Input parameters for the {@link contextExchange}. */

+5 -16

exchanges/execute/src/execute.ts

reviewed

··· 1 1 - import { 2 2 - Source, 3 3 - pipe, 4 4 - share, 5 5 - filter, 6 6 - takeUntil, 7 7 - mergeMap, 8 8 - merge, 9 9 - make, 10 10 - } from 'wonka'; 1 1 + import { Source, pipe, filter, takeUntil, mergeMap, merge, make } from 'wonka'; 11 2 12 3 import { 13 4 GraphQLSchema, ··· 131 122 export const executeExchange = 132 123 (options: ExecuteExchangeArgs): Exchange => 133 124 ({ forward }) => { 134 134 - return ops$ => { 135 135 - const sharedOps$ = share(ops$); 136 136 - 125 125 + return operations$ => { 137 126 const executedOps$ = pipe( 138 138 - sharedOps$, 127 127 + operations$, 139 128 filter((operation: Operation) => { 140 129 return ( 141 130 operation.kind === 'query' || ··· 146 135 mergeMap((operation: Operation) => { 147 136 const { key } = operation; 148 137 const teardown$ = pipe( 149 149 - sharedOps$, 138 138 + operations$, 150 139 filter(op => op.kind === 'teardown' && op.key === key) 151 140 ); 152 141 ··· 192 181 ); 193 182 194 183 const forwardedOps$ = pipe( 195 195 - sharedOps$, 184 184 + operations$, 196 185 filter(operation => operation.kind === 'teardown'), 197 186 forward 198 187 );

+83 -59

exchanges/graphcache/src/cacheExchange.test.ts

reviewed

··· 12 12 import { 13 13 Source, 14 14 pipe, 15 15 + share, 15 16 map, 16 17 merge, 17 18 mergeMap, ··· 86 87 87 88 const { source: ops$, next } = makeSubject<Operation>(); 88 89 const result = vi.fn(); 89 89 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 90 90 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 90 91 91 92 pipe( 92 93 cacheExchange({})({ forward, client, dispatchDebug })(ops$), ··· 137 138 138 139 const { source: ops$, next } = makeSubject<Operation>(); 139 140 const result = vi.fn(); 140 140 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 141 141 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 141 142 142 143 pipe( 143 144 cacheExchange({})({ forward, client, dispatchDebug })(ops$), ··· 214 215 return undefined as any; 215 216 }); 216 217 217 217 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 218 218 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 218 219 const result = vi.fn(); 219 220 220 221 pipe( ··· 358 359 }); 359 360 360 361 const result = vi.fn(); 361 361 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 362 362 + const forward: ExchangeIO = ops$ => 363 363 + pipe(ops$, delay(1), map(response), share); 362 364 363 365 const updates = { 364 366 Mutation: { ··· 459 461 return undefined as any; 460 462 }); 461 463 462 462 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 464 464 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 463 465 const result = vi.fn(); 464 466 465 467 pipe( ··· 519 521 }); 520 522 521 523 const result = vi.fn(); 522 522 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 524 524 + const forward: ExchangeIO = ops$ => 525 525 + pipe(ops$, delay(1), map(response), share); 523 526 524 527 const updates = { 525 528 Mutation: { ··· 581 584 }); 582 585 583 586 const result = vi.fn(); 584 584 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 587 587 + const forward: ExchangeIO = ops$ => 588 588 + pipe(ops$, delay(1), map(response), share); 585 589 586 590 const updates = { 587 591 Mutation: { ··· 661 665 }); 662 666 663 667 const result = vi.fn(); 664 664 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 668 668 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 665 669 666 670 pipe( 667 671 cacheExchange({})({ forward, client, dispatchDebug })(ops$), ··· 748 752 }); 749 753 750 754 const result = vi.fn(); 751 751 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 755 755 + const forward: ExchangeIO = ops$ => 756 756 + pipe(ops$, delay(1), map(response), share); 752 757 753 758 const optimistic = { 754 759 concealAuthor: vi.fn(() => optimisticMutationData.concealAuthor) as any, ··· 858 863 }); 859 864 860 865 const result = vi.fn(); 861 861 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(3), map(response)); 866 866 + const forward: ExchangeIO = ops$ => 867 867 + pipe(ops$, delay(3), map(response), share); 862 868 863 869 const optimistic = { 864 870 concealAuthor: vi.fn(() => optimisticMutationData.concealAuthor) as any, ··· 957 963 ops$, 958 964 delay(1), 959 965 filter(x => x.kind !== 'mutation'), 960 960 - map(response) 966 966 + map(response), 967 967 + share 961 968 ); 962 969 963 970 const optimistic = { ··· 1071 1078 }); 1072 1079 1073 1080 const result = vi.fn(); 1074 1074 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 1081 1081 + const forward: ExchangeIO = ops$ => 1082 1082 + pipe(ops$, delay(1), map(response), share); 1075 1083 1076 1084 const optimistic = { 1077 1085 addAuthor: vi.fn(() => optimisticMutationData.addAuthor) as any, ··· 1138 1146 return undefined as any; 1139 1147 }); 1140 1148 1141 1141 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 1149 1149 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 1142 1150 1143 1151 const result = vi.fn(); 1144 1152 const fakeResolver = vi.fn(); ··· 1225 1233 }); 1226 1234 1227 1235 const result = vi.fn(); 1228 1228 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 1236 1236 + const forward: ExchangeIO = ops$ => 1237 1237 + pipe(ops$, delay(1), map(response), share); 1229 1238 1230 1239 const fakeResolver = vi.fn(); 1231 1240 ··· 1377 1386 }); 1378 1387 1379 1388 const result = vi.fn(); 1380 1380 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 1389 1389 + const forward: ExchangeIO = ops$ => 1390 1390 + pipe(ops$, delay(1), map(response), share); 1381 1391 1382 1392 const fakeResolver = vi.fn(); 1383 1393 const called: any[] = []; ··· 1540 1550 }); 1541 1551 1542 1552 const result = vi.fn(); 1543 1543 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 1553 1553 + const forward: ExchangeIO = ops$ => 1554 1554 + pipe(ops$, delay(1), map(response), share); 1544 1555 1545 1556 pipe( 1546 1557 cacheExchange({ ··· 1683 1694 }); 1684 1695 1685 1696 const result = vi.fn(); 1686 1686 - const forward: ExchangeIO = ops$ => pipe(ops$, delay(1), map(response)); 1697 1697 + const forward: ExchangeIO = ops$ => 1698 1698 + pipe(ops$, delay(1), map(response), share); 1687 1699 1688 1700 pipe( 1689 1701 cacheExchange({ ··· 1859 1871 `; 1860 1872 1861 1873 const forward = (ops$: Source<Operation>): Source<OperationResult> => 1862 1862 - merge([ 1863 1863 - pipe( 1864 1864 - ops$, 1865 1865 - filter(() => false) 1866 1866 - ) as any, 1867 1867 - res$, 1868 1868 - ]); 1874 1874 + share( 1875 1875 + merge([ 1876 1876 + pipe( 1877 1877 + ops$, 1878 1878 + filter(() => false) 1879 1879 + ) as any, 1880 1880 + res$, 1881 1881 + ]) 1882 1882 + ); 1869 1883 1870 1884 const optimistic = { 1871 1885 node: () => ({ ··· 1953 1967 `; 1954 1968 1955 1969 const forward = (ops$: Source<Operation>): Source<OperationResult> => 1956 1956 - merge([ 1957 1957 - pipe( 1958 1958 - ops$, 1959 1959 - filter(() => false) 1960 1960 - ) as any, 1961 1961 - res$, 1962 1962 - ]); 1970 1970 + share( 1971 1971 + merge([ 1972 1972 + pipe( 1973 1973 + ops$, 1974 1974 + filter(() => false) 1975 1975 + ) as any, 1976 1976 + res$, 1977 1977 + ]) 1978 1978 + ); 1963 1979 1964 1980 pipe( 1965 1981 cacheExchange()({ forward, client, dispatchDebug })(ops$), ··· 2074 2090 `; 2075 2091 2076 2092 const forward = (ops$: Source<Operation>): Source<OperationResult> => 2077 2077 - merge([ 2078 2078 - pipe( 2079 2079 - ops$, 2080 2080 - filter(() => false) 2081 2081 - ) as any, 2082 2082 - res$, 2083 2083 - ]); 2093 2093 + share( 2094 2094 + merge([ 2095 2095 + pipe( 2096 2096 + ops$, 2097 2097 + filter(() => false) 2098 2098 + ) as any, 2099 2099 + res$, 2100 2100 + ]) 2101 2101 + ); 2084 2102 2085 2103 const optimistic = { 2086 2104 node: () => ({ ··· 2177 2195 `; 2178 2196 2179 2197 const forward = (ops$: Source<Operation>): Source<OperationResult> => 2180 2180 - merge([ 2181 2181 - pipe( 2182 2182 - ops$, 2183 2183 - filter(() => false) 2184 2184 - ) as any, 2185 2185 - res$, 2186 2186 - ]); 2198 2198 + share( 2199 2199 + merge([ 2200 2200 + pipe( 2201 2201 + ops$, 2202 2202 + filter(() => false) 2203 2203 + ) as any, 2204 2204 + res$, 2205 2205 + ]) 2206 2206 + ); 2187 2207 2188 2208 pipe( 2189 2209 cacheExchange()({ forward, client, dispatchDebug })(ops$), ··· 2286 2306 `; 2287 2307 2288 2308 const forward = (ops$: Source<Operation>): Source<OperationResult> => 2289 2289 - merge([ 2290 2290 - pipe( 2291 2291 - ops$, 2292 2292 - filter(() => false) 2293 2293 - ) as any, 2294 2294 - res$, 2295 2295 - ]); 2309 2309 + share( 2310 2310 + merge([ 2311 2311 + pipe( 2312 2312 + ops$, 2313 2313 + filter(() => false) 2314 2314 + ) as any, 2315 2315 + res$, 2316 2316 + ]) 2317 2317 + ); 2296 2318 2297 2319 pipe( 2298 2320 cacheExchange()({ forward, client, dispatchDebug })(ops$), ··· 2438 2460 `; 2439 2461 2440 2462 const forward = (ops$: Source<Operation>): Source<OperationResult> => 2441 2441 - merge([ 2442 2442 - pipe( 2443 2443 - ops$, 2444 2444 - filter(() => false) 2445 2445 - ) as any, 2446 2446 - res$, 2447 2447 - ]); 2463 2463 + share( 2464 2464 + merge([ 2465 2465 + pipe( 2466 2466 + ops$, 2467 2467 + filter(() => false) 2468 2468 + ) as any, 2469 2469 + res$, 2470 2470 + ]) 2471 2471 + ); 2448 2472 2449 2473 pipe( 2450 2474 cacheExchange()({ forward, client, dispatchDebug })(ops$),

+4 -7

exchanges/graphcache/src/cacheExchange.ts

reviewed

··· 295 295 }; 296 296 }; 297 297 298 298 - return ops$ => { 299 299 - const sharedOps$ = pipe(ops$, share); 300 300 - 298 298 + return operations$ => { 301 299 // Filter by operations that are cacheable and attempt to query them from the cache 302 300 const cacheOps$ = pipe( 303 303 - sharedOps$, 301 301 + operations$, 304 302 filter( 305 303 op => 306 304 op.kind === 'query' && op.context.requestPolicy !== 'network-only' ··· 310 308 ); 311 309 312 310 const nonCacheOps$ = pipe( 313 313 - sharedOps$, 311 311 + operations$, 314 312 filter( 315 313 op => 316 314 op.kind !== 'query' || op.context.requestPolicy === 'network-only' ··· 402 400 const result$ = pipe( 403 401 merge([nonCacheOps$, cacheMissOps$]), 404 402 map(prepareForwardedOperation), 405 405 - forward, 406 406 - share 403 403 + forward 407 404 ); 408 405 409 406 // Results that can immediately be resolved

+7 -7

exchanges/graphcache/src/offlineExchange.test.ts

reviewed

··· 8 8 import { print } from 'graphql'; 9 9 import { vi, expect, it, describe } from 'vitest'; 10 10 11 11 - import { pipe, map, makeSubject, tap, publish } from 'wonka'; 11 11 + import { pipe, share, map, makeSubject, tap, publish } from 'wonka'; 12 12 import { queryResponse } from '../../../packages/core/src/test-utils'; 13 13 import { offlineExchange } from './offlineExchange'; 14 14 ··· 88 88 89 89 const { source: ops$ } = makeSubject<Operation>(); 90 90 const result = vi.fn(); 91 91 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 91 91 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 92 92 93 93 vi.useFakeTimers(); 94 94 pipe( ··· 141 141 142 142 const { source: ops$, next } = makeSubject<Operation>(); 143 143 const result = vi.fn(); 144 144 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 144 144 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 145 145 146 146 pipe( 147 147 offlineExchange({ ··· 164 164 165 165 next(mutationOp); 166 166 expect(result).toBeCalledTimes(1); 167 167 - expect(storage.writeMetadata).toBeCalledTimes(1); 167 167 + expect(storage.writeMetadata).toHaveBeenCalled(); 168 168 expect(storage.writeMetadata).toHaveBeenCalledWith([ 169 169 { 170 170 query: `mutation { ··· 278 278 279 279 const { source: ops$, next } = makeSubject<Operation>(); 280 280 const result = vi.fn(); 281 281 - const forward: ExchangeIO = ops$ => pipe(ops$, map(response)); 281 281 + const forward: ExchangeIO = ops$ => pipe(ops$, map(response), share); 282 282 283 283 pipe( 284 284 offlineExchange({ ··· 296 296 ); 297 297 298 298 next(mutationOp); 299 299 - expect(storage.writeMetadata).toBeCalledTimes(1); 299 299 + expect(storage.writeMetadata).toHaveBeenCalled(); 300 300 expect(storage.writeMetadata).toHaveBeenCalledWith([ 301 301 { 302 302 query: `mutation { ··· 313 313 await onOnlineCalled; 314 314 315 315 flush!(); 316 316 - expect(reexecuteOperation).toHaveBeenCalledTimes(1); 316 316 + expect(reexecuteOperation).toHaveBeenCalled(); 317 317 expect((reexecuteOperation.mock.calls[0][0] as any).key).toEqual(1); 318 318 expect(print((reexecuteOperation.mock.calls[0][0] as any).query)).toEqual( 319 319 print(gql`

+3 -5

exchanges/graphcache/src/offlineExchange.ts

reviewed

··· 1 1 - import { pipe, merge, makeSubject, share, filter } from 'wonka'; 1 1 + import { pipe, merge, makeSubject, filter } from 'wonka'; 2 2 import { print, SelectionNode } from 'graphql'; 3 3 4 4 import { ··· 214 214 forward, 215 215 }); 216 216 217 217 - return ops$ => { 218 218 - const sharedOps$ = pipe(ops$, share); 219 219 - 220 220 - const opsAndRebound$ = merge([reboundOps$, sharedOps$]); 217 217 + return operations$ => { 218 218 + const opsAndRebound$ = merge([reboundOps$, operations$]); 221 219 222 220 return pipe( 223 221 cacheResults$(opsAndRebound$),

+5 -6

exchanges/multipart-fetch/src/multipartFetchExchange.ts

reviewed

··· 1 1 - import { filter, merge, mergeMap, pipe, share, takeUntil, onPush } from 'wonka'; 1 1 + import { filter, merge, mergeMap, pipe, takeUntil, onPush } from 'wonka'; 2 2 import { extractFiles } from 'extract-files'; 3 3 import { Exchange } from '@urql/core'; 4 4 ··· 15 15 */ 16 16 export const multipartFetchExchange: Exchange = 17 17 ({ forward, dispatchDebug }) => 18 18 - ops$ => { 19 19 - const sharedOps$ = share(ops$); 18 18 + operations$ => { 20 19 const fetchResults$ = pipe( 21 21 - sharedOps$, 20 20 + operations$, 22 21 filter(operation => { 23 22 return operation.kind === 'query' || operation.kind === 'mutation'; 24 23 }), 25 24 mergeMap(operation => { 26 25 const teardown$ = pipe( 27 27 - sharedOps$, 26 26 + operations$, 28 27 filter(op => op.kind === 'teardown' && op.key === operation.key) 29 28 ); 30 29 ··· 100 99 ); 101 100 102 101 const forward$ = pipe( 103 103 - sharedOps$, 102 102 + operations$, 104 103 filter(operation => { 105 104 return operation.kind !== 'query' && operation.kind !== 'mutation'; 106 105 }),

+2 -4

exchanges/persisted/src/persistedExchange.ts

reviewed

··· 6 6 merge, 7 7 mergeMap, 8 8 pipe, 9 9 - share, 10 9 } from 'wonka'; 11 10 12 11 import { ··· 131 130 132 131 return operations$ => { 133 132 const retries = makeSubject<Operation>(); 134 134 - const sharedOps$ = share(operations$); 135 133 136 134 const forwardedOps$ = pipe( 137 137 - sharedOps$, 135 135 + operations$, 138 136 filter(operation => !operationFilter(operation)) 139 137 ); 140 138 141 139 const persistedOps$ = pipe( 142 142 - sharedOps$, 140 140 + operations$, 143 141 filter(operationFilter), 144 142 map(async operation => { 145 143 const persistedOperation = makeOperation(operation.kind, operation, {

+6 -18

exchanges/retry/src/retryExchange.ts

reviewed

··· 1 1 import { 2 2 - Source, 3 2 makeSubject, 4 4 - share, 5 3 pipe, 6 4 merge, 7 5 filter, ··· 11 9 takeUntil, 12 10 } from 'wonka'; 13 11 14 14 - import { 15 15 - makeOperation, 16 16 - Exchange, 17 17 - Operation, 18 18 - CombinedError, 19 19 - OperationResult, 20 20 - } from '@urql/core'; 12 12 + import { makeOperation, Exchange, Operation, CombinedError } from '@urql/core'; 21 13 22 14 /** Input parameters for the {@link retryExchange}. */ 23 15 export interface RetryExchangeOptions { ··· 119 111 options.randomDelay != null ? !!options.randomDelay : true; 120 112 121 113 return ({ forward, dispatchDebug }) => 122 122 - ops$ => { 123 123 - const sharedOps$ = pipe(ops$, share); 114 114 + operations$ => { 124 115 const { source: retry$, next: nextRetryOperation } = 125 116 makeSubject<Operation>(); 126 117 ··· 142 133 // But if this event comes through regularly we also stop the retries, since it's 143 134 // basically the query retrying itself, no backoff should be added! 144 135 const teardown$ = pipe( 145 145 - sharedOps$, 136 136 + operations$, 146 137 filter(op => { 147 138 return ( 148 139 (op.kind === 'query' || op.kind === 'teardown') && ··· 176 167 }) 177 168 ); 178 169 179 179 - const result$ = pipe( 180 180 - merge([sharedOps$, retryWithBackoff$]), 170 170 + return pipe( 171 171 + merge([operations$, retryWithBackoff$]), 181 172 forward, 182 182 - share, 183 173 filter(res => { 184 174 // Only retry if the error passes the conditional retryIf function (if passed) 185 175 // or if the error contains a networkError ··· 216 206 217 207 return true; 218 208 }) 219 219 - ) as Source<OperationResult>; 220 220 - 221 221 - return result$; 209 209 + ); 222 210 }; 223 211 };

+4 -6

packages/core/src/exchanges/cache.ts

reviewed

··· 1 1 /* eslint-disable @typescript-eslint/no-use-before-define */ 2 2 - import { filter, map, merge, pipe, share, tap } from 'wonka'; 2 2 + import { filter, map, merge, pipe, tap } from 'wonka'; 3 3 4 4 import { Client } from '../client'; 5 5 import { Exchange, Operation, OperationResult } from '../types'; ··· 55 55 resultCache.has(operation.key)); 56 56 57 57 return ops$ => { 58 58 - const sharedOps$ = share(ops$); 59 59 - 60 58 const cachedOps$ = pipe( 61 61 - sharedOps$, 59 59 + ops$, 62 60 filter(op => !shouldSkip(op) && isOperationCached(op)), 63 61 map(operation => { 64 62 const cachedResult = resultCache.get(operation.key); ··· 98 96 const forwardedOps$ = pipe( 99 97 merge([ 100 98 pipe( 101 101 - sharedOps$, 99 99 + ops$, 102 100 filter(op => !shouldSkip(op) && !isOperationCached(op)), 103 101 map(mapTypeNames) 104 102 ), 105 103 pipe( 106 106 - sharedOps$, 104 104 + ops$, 107 105 filter(op => shouldSkip(op)) 108 106 ), 109 107 ]),

+29 -15

packages/core/src/exchanges/compose.ts

reviewed

··· 1 1 + import { share } from 'wonka'; 1 2 import type { ExchangeIO, Exchange, ExchangeInput } from '../types'; 2 3 3 4 /** Composes an array of Exchanges into a single one. ··· 13 14 * 14 15 * This simply merges all exchanges into one and is used by the {@link Client} 15 16 * to merge the `exchanges` option it receives. 17 17 + * 18 18 + * @throws 19 19 + * In development, if {@link ExchangeInput.forward} is called repeatedly 20 20 + * by an {@link Exchange} an error is thrown, since `forward()` must only 21 21 + * be called once per `Exchange`. 16 22 */ 17 23 export const composeExchanges = 18 24 (exchanges: Exchange[]): Exchange => 19 25 ({ client, forward, dispatchDebug }: ExchangeInput): ExchangeIO => 20 20 - exchanges.reduceRight( 21 21 - (forward, exchange) => 22 22 - exchange({ 23 23 - client, 24 24 - forward, 25 25 - dispatchDebug(event) { 26 26 - dispatchDebug({ 27 27 - timestamp: Date.now(), 28 28 - source: exchange.name, 29 29 - ...event, 30 30 - }); 31 31 - }, 32 32 - }), 33 33 - forward 34 34 - ); 26 26 + exchanges.reduceRight((forward, exchange) => { 27 27 + let forwarded = false; 28 28 + return exchange({ 29 29 + client, 30 30 + forward(operations$) { 31 31 + if (process.env.NODE_ENV !== 'production') { 32 32 + if (forwarded) 33 33 + throw new Error( 34 34 + 'forward() must only be called once in each Exchange.' 35 35 + ); 36 36 + forwarded = true; 37 37 + } 38 38 + return share(forward(share(operations$))); 39 39 + }, 40 40 + dispatchDebug(event) { 41 41 + dispatchDebug({ 42 42 + timestamp: Date.now(), 43 43 + source: exchange.name, 44 44 + ...event, 45 45 + }); 46 46 + }, 47 47 + }); 48 48 + }, forward);

+4 -5

packages/core/src/exchanges/fetch.ts

reviewed

··· 1 1 /* eslint-disable @typescript-eslint/no-use-before-define */ 2 2 - import { filter, merge, mergeMap, pipe, share, takeUntil, onPush } from 'wonka'; 2 2 + import { filter, merge, mergeMap, pipe, takeUntil, onPush } from 'wonka'; 3 3 4 4 import { Exchange } from '../types'; 5 5 import { ··· 28 28 */ 29 29 export const fetchExchange: Exchange = ({ forward, dispatchDebug }) => { 30 30 return ops$ => { 31 31 - const sharedOps$ = share(ops$); 32 31 const fetchResults$ = pipe( 33 33 - sharedOps$, 32 32 + ops$, 34 33 filter(operation => { 35 34 return operation.kind === 'query' || operation.kind === 'mutation'; 36 35 }), ··· 53 52 makeFetchSource(operation, url, fetchOptions), 54 53 takeUntil( 55 54 pipe( 56 56 - sharedOps$, 55 55 + ops$, 57 56 filter(op => op.kind === 'teardown' && op.key === operation.key) 58 57 ) 59 58 ) ··· 86 85 ); 87 86 88 87 const forward$ = pipe( 89 89 - sharedOps$, 88 88 + ops$, 90 89 filter(operation => { 91 90 return operation.kind !== 'query' && operation.kind !== 'mutation'; 92 91 }),

+3 -5

packages/core/src/exchanges/ssr.ts

reviewed

··· 1 1 import { GraphQLError } from 'graphql'; 2 2 - import { pipe, share, filter, merge, map, tap } from 'wonka'; 2 2 + import { pipe, filter, merge, map, tap } from 'wonka'; 3 3 import { Exchange, OperationResult, Operation } from '../types'; 4 4 import { addMetadata, CombinedError } from '../utils'; 5 5 import { reexecuteOperation } from './cache'; ··· 218 218 ? !!params.isClient 219 219 : !client.suspense; 220 220 221 221 - const sharedOps$ = share(ops$); 222 222 - 223 221 let forwardedOps$ = pipe( 224 224 - sharedOps$, 222 222 + ops$, 225 223 filter( 226 224 operation => 227 225 !data[operation.key] || ··· 234 232 // NOTE: Since below we might delete the cached entry after accessing 235 233 // it once, cachedOps$ needs to be merged after forwardedOps$ 236 234 let cachedOps$ = pipe( 237 237 - sharedOps$, 235 235 + ops$, 238 236 filter( 239 237 operation => 240 238 !!data[operation.key] &&

+3 -5

packages/core/src/exchanges/subscription.ts

reviewed

··· 4 4 merge, 5 5 mergeMap, 6 6 pipe, 7 7 - share, 8 7 Subscription, 9 8 Source, 10 9 takeUntil, ··· 198 197 }); 199 198 200 199 return ops$ => { 201 201 - const sharedOps$ = share(ops$); 202 200 const subscriptionResults$ = pipe( 203 203 - sharedOps$, 201 201 + ops$, 204 202 filter(isSubscriptionOperationFn), 205 203 mergeMap(operation => { 206 204 const { key } = operation; 207 205 const teardown$ = pipe( 208 208 - sharedOps$, 206 206 + ops$, 209 207 filter(op => op.kind === 'teardown' && op.key === key) 210 208 ); 211 209 ··· 217 215 ); 218 216 219 217 const forward$ = pipe( 220 220 - sharedOps$, 218 218 + ops$, 221 219 filter(op => !isSubscriptionOperationFn(op)), 222 220 forward 223 221 );

+15

packages/core/src/types.ts

reviewed

··· 676 676 * 677 677 * @see {@link https://urql.dev/goto/docs/architecture/#the-client-and-exchanges} for more information on Exchanges. 678 678 * @see {@link https://urql.dev/goto/docs/advanced/authoring-exchanges} on how Exchanges are authored. 679 679 + * 680 680 + * @example 681 681 + * ```ts 682 682 + * import { pipe, onPush } from 'wonka'; 683 683 + * import { Exchange } from '@urql/core'; 684 684 + * 685 685 + * const debugExchange: Exchange => { 686 686 + * return ops$ => pipe( 687 687 + * ops$, 688 688 + * onPush(operation => console.log(operation)), 689 689 + * forward, 690 690 + * onPush(result => console.log(result)), 691 691 + * ); 692 692 + * }; 693 693 + * ``` 679 694 */ 680 695 export type Exchange = (input: ExchangeInput) => ExchangeIO; 681 696