Simplify type params where they aren't needed
Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>
This commit is contained in:
+3
-5
@@ -23,7 +23,7 @@ import { v4 } from 'uuid';
|
||||
import { stringifyError } from '@backstage/errors';
|
||||
import { EventParams, EventSubscriber } from '@backstage/plugin-events-node';
|
||||
|
||||
export class IncrementalIngestionEngine<TInput>
|
||||
export class IncrementalIngestionEngine
|
||||
implements IterationEngine, EventSubscriber
|
||||
{
|
||||
private readonly restLength: Duration;
|
||||
@@ -32,7 +32,7 @@ export class IncrementalIngestionEngine<TInput>
|
||||
|
||||
private manager: IncrementalIngestionDatabaseManager;
|
||||
|
||||
constructor(private options: IterationEngineOptions<TInput>) {
|
||||
constructor(private options: IterationEngineOptions) {
|
||||
this.manager = options.manager;
|
||||
this.restLength = Duration.fromObject(options.restLength);
|
||||
this.backoff = options.backoff ?? [
|
||||
@@ -347,13 +347,11 @@ export class IncrementalIngestionEngine<TInput>
|
||||
`incremental-engine: Received ${this.providerEventTopic} event`,
|
||||
);
|
||||
|
||||
const payload = eventPayload as TInput;
|
||||
|
||||
if (!provider.deltaMapper) {
|
||||
return;
|
||||
}
|
||||
|
||||
const update = provider.deltaMapper(payload);
|
||||
const update = provider.deltaMapper(eventPayload);
|
||||
|
||||
if (update.delta) {
|
||||
if (update.delta.added.length > 0) {
|
||||
|
||||
+2
-2
@@ -41,7 +41,7 @@ describe('WrapperProviders', () => {
|
||||
async databaseId => {
|
||||
const client = await databases.init(databaseId);
|
||||
|
||||
const provider1: IncrementalEntityProvider<number, {}> = {
|
||||
const provider1: IncrementalEntityProvider = {
|
||||
getProviderName: () => 'provider1',
|
||||
around: burst => burst(0),
|
||||
next: async (_context, cursor) => {
|
||||
@@ -51,7 +51,7 @@ describe('WrapperProviders', () => {
|
||||
},
|
||||
};
|
||||
|
||||
const provider2: IncrementalEntityProvider<number, {}> = {
|
||||
const provider2: IncrementalEntityProvider = {
|
||||
getProviderName: () => 'provider2',
|
||||
around: burst => burst(0),
|
||||
next: async (_context, cursor) => {
|
||||
|
||||
@@ -58,7 +58,7 @@ export class WrapperProviders {
|
||||
) {}
|
||||
|
||||
wrap(
|
||||
provider: IncrementalEntityProvider<unknown, unknown>,
|
||||
provider: IncrementalEntityProvider,
|
||||
options: IncrementalEntityProviderOptions,
|
||||
): EntityProvider {
|
||||
this.numberOfProvidersToConnect += 1;
|
||||
@@ -81,8 +81,8 @@ export class WrapperProviders {
|
||||
).createRouter();
|
||||
}
|
||||
|
||||
private async startProvider<TCursor, TContext, TInput>(
|
||||
provider: IncrementalEntityProvider<TCursor, TContext, TInput>,
|
||||
private async startProvider(
|
||||
provider: IncrementalEntityProvider,
|
||||
providerOptions: IncrementalEntityProviderOptions,
|
||||
connection: EntityProviderConnection,
|
||||
) {
|
||||
|
||||
+1
-1
@@ -24,7 +24,7 @@ import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIn
|
||||
|
||||
describe('bitbucketServerEntityProviderCatalogModule', () => {
|
||||
it('should register provider at the catalog extension point', async () => {
|
||||
const provider1: IncrementalEntityProvider<number, {}, unknown> = {
|
||||
const provider1: IncrementalEntityProvider = {
|
||||
getProviderName: () => 'provider1',
|
||||
around: burst => burst(0),
|
||||
next: async (cursor, _context) => {
|
||||
|
||||
+1
-1
@@ -38,7 +38,7 @@ export const incrementalIngestionEntityProviderCatalogModule =
|
||||
env,
|
||||
options: {
|
||||
providers: Array<{
|
||||
provider: IncrementalEntityProvider<unknown, unknown, unknown>;
|
||||
provider: IncrementalEntityProvider;
|
||||
options: IncrementalEntityProviderOptions;
|
||||
}>;
|
||||
},
|
||||
|
||||
@@ -36,10 +36,10 @@ import {
|
||||
incrementalIngestionEntityProviderCatalogModule,
|
||||
} from '.';
|
||||
|
||||
const provider: IncrementalEntityProvider<number, {}, unknown> = {
|
||||
const provider: IncrementalEntityProvider = {
|
||||
getProviderName: () => 'test-provider',
|
||||
around: burst => burst(0),
|
||||
next: async (_context, cursor) => {
|
||||
next: async (_context, cursor: number | undefined) => {
|
||||
await new Promise(resolve => setTimeout(resolve, 500));
|
||||
if (cursor === undefined || cursor < 3) {
|
||||
console.log(`### Returning batch #${cursor}`);
|
||||
|
||||
+2
-2
@@ -68,8 +68,8 @@ export class IncrementalCatalogBuilder {
|
||||
return { incrementalAdminRouter };
|
||||
}
|
||||
|
||||
addIncrementalEntityProvider<TCursor, TContext, TInput>(
|
||||
provider: IncrementalEntityProvider<TCursor, TContext, TInput>,
|
||||
addIncrementalEntityProvider(
|
||||
provider: IncrementalEntityProvider,
|
||||
options: IncrementalEntityProviderOptions,
|
||||
) {
|
||||
const { burstInterval, burstLength, restLength } = options;
|
||||
|
||||
@@ -46,7 +46,7 @@ import { IncrementalIngestionDatabaseManager } from './database/IncrementalInges
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export interface IncrementalEntityProvider<TCursor, TContext, TInput = null> {
|
||||
export interface IncrementalEntityProvider {
|
||||
/**
|
||||
* This name must be unique between all of the entity providers
|
||||
* operating in the catalog.
|
||||
@@ -62,10 +62,7 @@ export interface IncrementalEntityProvider<TCursor, TContext, TInput = null> {
|
||||
* @returns The entities to be ingested, as well as the cursor of
|
||||
* the next page after this one.
|
||||
*/
|
||||
next(
|
||||
context: TContext,
|
||||
cursor?: TCursor,
|
||||
): Promise<EntityIteratorResult<TCursor>>;
|
||||
next(context: unknown, cursor?: unknown): Promise<EntityIteratorResult>;
|
||||
|
||||
/**
|
||||
* Do any setup and teardown necessary in order to provide the
|
||||
@@ -74,13 +71,13 @@ export interface IncrementalEntityProvider<TCursor, TContext, TInput = null> {
|
||||
*
|
||||
* @param burst - a function which performs a series of iterations
|
||||
*/
|
||||
around(burst: (context: TContext) => Promise<void>): Promise<void>;
|
||||
around(burst: (context: unknown) => Promise<void>): Promise<void>;
|
||||
|
||||
/**
|
||||
* If present, this method maps incoming payloads to apply updates
|
||||
* outside of the incremental ingestion schedule.
|
||||
*/
|
||||
deltaMapper?: (payload: TInput) => {
|
||||
deltaMapper?: (payload: unknown) => {
|
||||
delta:
|
||||
| {
|
||||
added: DeferredEntity[];
|
||||
@@ -96,16 +93,16 @@ export interface IncrementalEntityProvider<TCursor, TContext, TInput = null> {
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export type EntityIteratorResult<T> =
|
||||
export type EntityIteratorResult =
|
||||
| {
|
||||
done: false;
|
||||
entities: DeferredEntity[];
|
||||
cursor: T;
|
||||
cursor: unknown;
|
||||
}
|
||||
| {
|
||||
done: true;
|
||||
entities?: DeferredEntity[];
|
||||
cursor?: T;
|
||||
cursor?: unknown;
|
||||
};
|
||||
|
||||
/** @public */
|
||||
@@ -167,11 +164,11 @@ export interface IterationEngine {
|
||||
taskFn: TaskFunction;
|
||||
}
|
||||
|
||||
export interface IterationEngineOptions<TInput> {
|
||||
export interface IterationEngineOptions {
|
||||
logger: Logger;
|
||||
connection: EntityProviderConnection;
|
||||
manager: IncrementalIngestionDatabaseManager;
|
||||
provider: IncrementalEntityProvider<unknown, unknown, TInput>;
|
||||
provider: IncrementalEntityProvider;
|
||||
restLength: DurationObjectUnits;
|
||||
ready: Promise<void>;
|
||||
backoff?: IncrementalEntityProviderOptions['backoff'];
|
||||
|
||||
Reference in New Issue
Block a user